fix(langgraph): keep only name, description and forked in __subagents

`interrupt()` inside a tool raises `GraphInterrupt`, and langchain-core
calls `on_tool_error` for every exception, so the `tools` stream
reported a tool that was only waiting for an answer as `tool-error`,
with the interrupt flattened into the message. With a subagent running
inside a tool, the outer tool showed as failed while the subagent waited
for approval. Fixes #8218.

## Fix

- An interrupt or a drain ends the tool's events without a `tool-error`.
On resume the tool runs again and emits `tool-started` and
`tool-finished`. A parent command doesn't rerun the tool, so it still
ends the call with `tool-error`, as before.
- `astream` no longer runs every node in streaming mode just because the
`tools` handler is attached. Like the `messages` handler, it doesn't
need it, and with it a `@task` awaited inside a subgraph node returned
`None` whenever the `tools` mode was on.

The new tests fail on `main`.

## Release note

The `tools` stream no longer sends `tool-error` for a tool that pauses
on `interrupt()`. Until the run resumes, that tool has a `tool-started`
event and no end event. In-process `stream_events(version="v3")` now
ends that tool's stream as completed with no output, instead of failed
with the interrupt.

Same idea as #8219 by @haseeb-m4235 and #8282 by @thsfranca.

libs/langgraph/langgraph/pregel/_draw.py

@@ -228,6 +228,12 @@ def draw_graph(
             metadata["__interrupt"] = "before"
         elif name in interrupt_after_nodes:
             metadata["__interrupt"] = "after"
+        # Duck-typed so core doesn't import ToolNode from langgraph-prebuilt.
+        tools_by_name = getattr(node.bound, "tools_by_name", None)
+        if isinstance(tools_by_name, Mapping) and tools_by_name:
+            metadata["__tools"] = list(tools_by_name)
+            if subagents := _get_tool_subagents(tools_by_name):
+                metadata["__subagents"] = subagents
         graph.add_node(node.bound, name, metadata=metadata or None)
     # add start node
     if START not in nodes:
@@ -277,6 +283,38 @@ def draw_graph(
     return graph
 
 
+def _get_tool_subagents(
+    tools_by_name: Mapping[str, Any],
+) -> dict[str, list[dict[str, Any]]]:
+    """Collect the subagents each tool can run, keyed by tool name."""
+    subagents: dict[str, list[dict[str, Any]]] = {}
+    for tool_name, tool in tools_by_name.items():
+        tool_subagents = getattr(tool, "subagents", None)
+        if (
+            isinstance(tool_subagents, Sequence)
+            and not isinstance(tool_subagents, str)
+            and tool_subagents
+            and all(
+                isinstance(subagent, Mapping) and isinstance(subagent.get("name"), str)
+                for subagent in tool_subagents
+            )
+        ):
+            subagents[tool_name] = [
+                _subagent_entry(subagent) for subagent in tool_subagents
+            ]
+    return subagents
+
+
+def _subagent_entry(subagent: Mapping[str, Any]) -> dict[str, Any]:
+    """Keep only the subagent fields graph viewers display."""
+    entry: dict[str, Any] = {"name": subagent["name"]}
+    if isinstance(description := subagent.get("description"), str):
+        entry["description"] = description
+    if subagent.get("forked") is True:
+        entry["forked"] = True
+    return entry
+
+
 def add_edge(
     graph: Graph,
     source: str,

libs/langgraph/langgraph/pregel/_retry.py

@@ -34,6 +34,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_THREAD_ID,
     CONFIG_KEY_TIMED_ATTEMPT_OBSERVER,
     NS_SEP,
+    RETURN,
 )
 from langgraph._internal._runnable import create_task_in_config_context
 from langgraph._internal._timeout import sync_timeout_unsupported
@@ -733,15 +734,14 @@ async def arun_with_retry(
                 if stream:
                     async for _ in task.proc.astream(task.input, config):
                         pass
-                    break
+                    return _return_value(task)
                 return await task.proc.ainvoke(task.input, config)
             result = await _arun_with_timeout(
                 task, config, resolved_timeout, attempt_ctx, stream=stream
             )
             _finish_timed_attempt(config, attempt_ctx)
             if stream:
-                # if successful, end
-                break
+                return _return_value(task)
             return result
         except ParentCommand as exc:
             ns: str = config[CONF][CONFIG_KEY_CHECKPOINT_NS]
@@ -831,6 +831,12 @@ async def arun_with_retry(
             config = patch_configurable(config, {CONFIG_KEY_RESUMING: True})
 
 
+def _return_value(task: PregelExecutableTask) -> Any:
+    # Streaming drains the task instead of returning its result, so a caller
+    # awaiting a `@task` needs the value the task wrote.
+    return next((value for channel, value in task.writes if channel == RETURN), None)
+
+
 def _should_retry_on(retry_policy: RetryPolicy, exc: Exception) -> bool:
     """Check if the given exception should be retried based on the retry policy."""
     if isinstance(retry_policy.retry_on, Sequence):

libs/langgraph/langgraph/pregel/_tools.py

@@ -9,6 +9,7 @@ from langchain_core.callbacks import BaseCallbackHandler
 
 from langgraph._internal._constants import NS_SEP
 from langgraph.constants import TAG_NOSTREAM
+from langgraph.errors import GraphDrained, GraphInterrupt
 from langgraph.pregel.protocol import StreamChunk
 
 try:
@@ -188,6 +189,10 @@ class StreamToolCallHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             return
         ns, tool_call_id, token = info
         self._reset_writer(token)
+        # An interrupted or drained tool is paused, not failed: it runs again
+        # on resume and reports its start and end then.
+        if isinstance(error, (GraphInterrupt, GraphDrained)):
+            return
         self.stream(
             (
                 ns,

libs/langgraph/langgraph/pregel/main.py

@@ -3310,7 +3310,9 @@ class Pregel(
                         True
                         for h in run_manager.handlers
                         if isinstance(h, _StreamingCallbackHandler)
-                        and not isinstance(h, StreamMessagesHandler)
+                        and not isinstance(
+                            h, (StreamMessagesHandler, StreamToolCallHandler)
+                        )
                     ),
                     False,
                 )

libs/langgraph/tests/__snapshots__/test_large_cases.ambr

@@ -133,6 +133,11 @@
             "ToolNode"
           ],
           "name": "tools"
+        },
+        "metadata": {
+          "__tools": [
+            "search_api"
+          ]
         }
       },
       {
@@ -221,6 +226,11 @@
             "ToolNode"
           ],
           "name": "tools"
+        },
+        "metadata": {
+          "__tools": [
+            "search_api"
+          ]
         }
       },
       {

libs/langgraph/tests/test_pregel.py

@@ -36,7 +36,7 @@ from langchain_core.runnables import (
     RunnablePassthrough,
 )
 from langchain_core.runnables.graph import Edge
-from langchain_core.tools import tool
+from langchain_core.tools import StructuredTool, tool
 from langchain_core.version import VERSION as LANGCHAIN_CORE_VERSION
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
@@ -3105,6 +3105,49 @@ def test_simple_multi_edge(snapshot: SnapshotAssertion) -> None:
     )
 
 
+def test_tool_node_lists_tool_subagents_in_graph_metadata() -> None:
+    class TaskTool(StructuredTool):
+        subagents: list[dict[str, Any]] = []
+
+    def run_task(description: str, subagent_type: str) -> str:
+        return subagent_type
+
+    task = TaskTool.from_function(run_task, name="task", description="Run a subagent.")
+    task.subagents = [
+        {"name": "researcher", "description": "Researches a topic."},
+        {
+            "name": "critic",
+            "description": "Reviews drafts.",
+            "forked": True,
+            "system_prompt": "You are a critic.",
+            "runnable": object(),
+        },
+        {"name": "planner", "description": None, "forked": "yes"},
+    ]
+
+    @tool
+    def search(query: str) -> str:
+        """Search the web."""
+        return query
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("agent", lambda state: state)
+    builder.add_node("tools", ToolNode([search, task]))
+    builder.add_edge(START, "agent")
+    builder.add_edge("agent", "tools")
+    graph = builder.compile().get_graph().to_json()
+    nodes = {node["id"]: node for node in graph["nodes"]}
+
+    assert nodes["tools"]["metadata"]["__subagents"] == {
+        "task": [
+            {"name": "researcher", "description": "Researches a topic."},
+            {"name": "critic", "description": "Reviews drafts.", "forked": True},
+            {"name": "planner"},
+        ]
+    }
+    assert "__subagents" not in (nodes["agent"].get("metadata") or {})
+
+
 def test_nested_graph_xray(snapshot: SnapshotAssertion) -> None:
     class State(TypedDict):
         my_key: Annotated[str, operator.add]

libs/langgraph/tests/test_pregel_async.py

@@ -6834,6 +6834,31 @@ async def test_multiple_updates() -> None:
     ]
 
 
+@NEEDS_CONTEXTVARS
+async def test_task_in_subgraph_node_returns_its_value_with_tools_mode() -> None:
+    class State(TypedDict):
+        out: int | None
+
+    @task
+    async def double(x: int) -> int:
+        return x * 2
+
+    async def node(state: State) -> State:
+        return {"out": await double(21)}
+
+    inner = StateGraph(State).add_node("node", node).add_edge(START, "node").compile()
+    graph = (
+        StateGraph(State).add_node("inner", inner).add_edge(START, "inner").compile()
+    )
+
+    chunks = [
+        chunk
+        async for chunk in graph.astream({"out": None}, stream_mode=["values", "tools"])
+    ]
+
+    assert chunks[-1] == ("values", {"out": 42})
+
+
 @NEEDS_CONTEXTVARS
 async def test_falsy_return_from_task(async_checkpointer: BaseCheckpointSaver) -> None:
     """Test with a falsy return from a task."""
@@ -7598,6 +7623,40 @@ async def test_entrypoint_from_async_generator() -> None:
             yield "b"
 
 
+@NEEDS_CONTEXTVARS
+@pytest.mark.parametrize("timeout", [None, 30])
+@pytest.mark.parametrize("version", ["v1", "v2"])
+async def test_awaited_task_returns_its_value_under_astream_events(
+    version: Literal["v1", "v2"], timeout: int | None
+) -> None:
+    class State(TypedDict):
+        doubled: int
+        zeroed: int
+
+    @task(timeout=timeout)
+    async def double(x: int) -> int:
+        return x * 2
+
+    @task(timeout=timeout)
+    async def zero() -> int:
+        return 0
+
+    async def node(state: State) -> State:
+        return {"doubled": await double(21), "zeroed": await zero()}
+
+    graph = StateGraph(State).add_node("node", node).add_edge(START, "node").compile()
+
+    outputs = [
+        event["data"]["output"]
+        async for event in graph.astream_events(
+            {"doubled": 0, "zeroed": 1}, version=version
+        )
+        if event["event"] == "on_chain_end" and event["name"] == "node"
+    ]
+
+    assert outputs == [{"doubled": 42, "zeroed": 0}]
+
+
 @NEEDS_CONTEXTVARS
 async def test_named_tasks_functional() -> None:
     class Foo:

libs/langgraph/tests/test_tool_stream_handler.py

@@ -13,13 +13,17 @@ from typing import Annotated, Any
 import pytest
 from langchain_core.messages import AIMessage
 from langchain_core.tools import tool
+from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.prebuilt import ToolNode, ToolRuntime
 from typing_extensions import TypedDict
 
 from langgraph.constants import END, START
+from langgraph.errors import GraphDrained
 from langgraph.graph import StateGraph
 from langgraph.graph.message import add_messages
 from langgraph.pregel._tools import _tool_call_writer
+from langgraph.runtime import RunControl
+from langgraph.types import Command, interrupt
 
 
 class _State(TypedDict):
@@ -54,14 +58,14 @@ def _caller_async(tool_name: str, tool_args: dict[str, Any], tc_id: str = "tc1")
     return caller
 
 
-def _build_graph(caller, tools) -> Any:
+def _build_graph(caller, tools, checkpointer=None) -> Any:
     sg = StateGraph(_State)
     sg.add_node("caller", caller)
     sg.add_node("tools", ToolNode(tools))
     sg.add_edge(START, "caller")
     sg.add_edge("caller", "tools")
     sg.add_edge("tools", END)
-    return sg.compile()
+    return sg.compile(checkpointer=checkpointer)
 
 
 def _tool_events(stream) -> list[tuple[tuple[str, ...], dict]]:
@@ -146,6 +150,105 @@ class TestSyncGraphSyncTool:
         assert kinds == ["tool-started", "tool-error"]
         assert events[1][1]["message"] == "nope"
 
+    def test_interrupt_in_tool_is_not_a_tool_error(self) -> None:
+        @tool
+        def ask() -> str:
+            """asks."""
+            return interrupt("ok?")
+
+        graph = _build_graph(_caller_sync("ask", {}), [ask], InMemorySaver())
+        config = {"configurable": {"thread_id": "1"}}
+        paused = _tool_events(
+            graph.stream(
+                {"messages": []}, config, stream_mode=["tools"], subgraphs=True
+            )
+        )
+        resumed = _tool_events(
+            graph.stream(
+                Command(resume="yes"), config, stream_mode=["tools"], subgraphs=True
+            )
+        )
+
+        assert [p["event"] for _, p in paused] == ["tool-started"]
+        assert [(p["event"], p["tool_call_id"]) for _, p in resumed] == [
+            ("tool-started", "tc1"),
+            ("tool-finished", "tc1"),
+        ]
+
+    def test_tool_handing_off_to_the_parent_still_ends_its_call(self) -> None:
+        class State(TypedDict):
+            messages: Annotated[list, add_messages]
+            done: bool
+
+        def hand_off(state: State) -> Command:
+            return Command(graph=Command.PARENT, goto="finish", update={"done": True})
+
+        inner = StateGraph(State).add_node("hand_off", hand_off)
+        inner = inner.add_edge(START, "hand_off").compile()
+
+        @tool
+        def handoff() -> str:
+            """hands off."""
+            inner.invoke({"messages": [], "done": False})
+            return "unreachable"
+
+        graph = (
+            StateGraph(State)
+            .add_node("caller", _caller_sync("handoff", {}))
+            .add_node("tools", ToolNode([handoff]))
+            .add_node("finish", lambda state: {"done": True})
+            .add_edge(START, "caller")
+            .add_edge("caller", "tools")
+            .add_edge("tools", END)
+            .add_edge("finish", END)
+            .compile()
+        )
+        events = [
+            payload["event"]
+            for _, payload in _tool_events(
+                graph.stream(
+                    {"messages": [], "done": False},
+                    stream_mode=["tools"],
+                    subgraphs=True,
+                )
+            )
+        ]
+
+        assert events[0] == "tool-started"
+        assert events[-1] == "tool-error", events
+
+    def test_drained_tool_is_not_a_tool_error(self) -> None:
+        control = RunControl()
+
+        def request_drain(state: _State) -> dict:
+            control.request_drain()
+            return {}
+
+        inner = (
+            StateGraph(_State)
+            .add_node("request_drain", request_drain)
+            .add_node("after", lambda state: {})
+            .add_edge(START, "request_drain")
+            .add_edge("request_drain", "after")
+            .compile()
+        )
+
+        @tool
+        def drained() -> str:
+            """drains."""
+            inner.invoke({"messages": []}, control=control)
+            return "unreachable"
+
+        graph = _build_graph(_caller_sync("drained", {}), [drained])
+        chunks: list = []
+        with pytest.raises(GraphDrained):
+            for chunk in graph.stream(
+                {"messages": []}, stream_mode=["tools"], subgraphs=True
+            ):
+                chunks.append(chunk)
+
+        assert [p["event"] for _, p in _tool_events(chunks)] == ["tool-started"]
+
     def test_writer_unset_outside_tool(self) -> None:
         # Outside any tool body the ContextVar that ToolRuntime reads
         # is unset — emitting from there would be a no-op.
@@ -196,6 +299,25 @@ class TestAsyncGraphAsyncTool:
         assert kinds == ["tool-started", "tool-output-delta", "tool-finished"]
         assert events[1][1]["delta"] == "hi"
 
+    @pytest.mark.anyio
+    async def test_interrupt_in_async_tool_is_not_a_tool_error(self) -> None:
+        @tool
+        async def aask() -> str:
+            """asks."""
+            return interrupt("ok?")
+
+        graph = _build_graph(_caller_async("aask", {}), [aask], InMemorySaver())
+        config = {"configurable": {"thread_id": "1"}}
+        events = []
+        for graph_input in ({"messages": []}, Command(resume="yes")):
+            async for ns, mode, payload in graph.astream(
+                graph_input, config, stream_mode=["tools"], subgraphs=True
+            ):
+                if mode == "tools":
+                    events.append(payload["event"])
+
+        assert events == ["tool-started", "tool-started", "tool-finished"]
+
 
 class TestConcurrentToolCalls:
     def test_parallel_tool_calls_do_not_bleed(self) -> None:

libs/sdk-py/langgraph_sdk/_async/http.py

@@ -217,6 +217,12 @@ class HttpClient:
             current_headers = dict(
                 request_headers if reconnect_path is None else reconnect_headers
             )
+            if last_event_id is not None:
+                current_headers = {
+                    key: value
+                    for key, value in current_headers.items()
+                    if key.lower() != "last-event-id"
+                }
             if last_event_id:
                 current_headers["Last-Event-ID"] = last_event_id
 

libs/sdk-py/langgraph_sdk/_sync/http.py

@@ -219,6 +219,12 @@ class SyncHttpClient:
             current_headers = dict(
                 request_headers if reconnect_path is None else reconnect_headers
             )
+            if last_event_id is not None:
+                current_headers = {
+                    key: value
+                    for key, value in current_headers.items()
+                    if key.lower() != "last-event-id"
+                }
             if last_event_id:
                 current_headers["Last-Event-ID"] = last_event_id
 

libs/sdk-py/langgraph_sdk/sse.py

@@ -104,7 +104,7 @@ class SSEDecoder:
             if (
                 not self._event
                 and not self._data
-                and not self._last_event_id
+                and self._last_event_id is None
                 and self._retry is None
             ):
                 return None

libs/sdk-py/tests/test_client_stream.py

@@ -312,18 +312,14 @@ async def test_http_client_stream_recovers_after_disconnect():
     ]
 
 
-def _drop_after_an_empty_id(stream_cls: type, reconnects: list[httpx.Request]):
+def _drop_after(chunks: list[bytes], stream_cls: type, reconnects: list[httpx.Request]):
     def handler(request: httpx.Request) -> httpx.Response:
         if request.method == "POST":
             return httpx.Response(
                 200,
                 headers={"Content-Type": "text/event-stream", "Location": "/reconnect"},
                 stream=stream_cls(
-                    [
-                        b"id: 1\nevent: values\ndata: 1\n\n",
-                        b"id:\nevent: values\ndata: 2\n\n",
-                    ],
-                    httpx.RemoteProtocolError("incomplete chunked read"),
+                    chunks, httpx.RemoteProtocolError("incomplete chunked read")
                 ),
             )
         reconnects.append(request)
@@ -336,26 +332,99 @@ def _drop_after_an_empty_id(stream_cls: type, reconnects: list[httpx.Request]):
     return handler
 
 
-def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id():
+def _sync_reconnect(
+    chunks: list[bytes], headers: dict[str, str] | None
+) -> httpx.Request:
     reconnects: list[httpx.Request] = []
-    transport = httpx.MockTransport(_drop_after_an_empty_id(ListByteStream, reconnects))
+    transport = httpx.MockTransport(_drop_after(chunks, ListByteStream, reconnects))
     with httpx.Client(transport=transport, base_url="https://example.com") as client:
-        list(SyncHttpClient(client).stream("/stream", "POST", json={}))
-
-    assert "last-event-id" not in reconnects[0].headers
+        list(SyncHttpClient(client).stream("/stream", "POST", json={}, headers=headers))
+    return reconnects[0]
 
 
-async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id():
+async def _async_reconnect(
+    chunks: list[bytes], headers: dict[str, str] | None
+) -> httpx.Request:
     reconnects: list[httpx.Request] = []
     transport = httpx.MockTransport(
-        _drop_after_an_empty_id(AsyncListByteStream, reconnects)
+        _drop_after(chunks, AsyncListByteStream, reconnects)
     )
     async with httpx.AsyncClient(
         transport=transport, base_url="https://example.com"
     ) as client:
-        [part async for part in HttpClient(client).stream("/stream", "POST", json={})]
+        [
+            part
+            async for part in HttpClient(client).stream(
+                "/stream", "POST", json={}, headers=headers
+            )
+        ]
+    return reconnects[0]
 
-    assert "last-event-id" not in reconnects[0].headers
+
+_EMPTY_IDS = pytest.mark.parametrize(
+    "empty_id",
+    [b"id:\nevent: values\ndata: 2\n\n", b"id:\n\n"],
+    ids=["with-an-event", "alone"],
+)
+_CALLER_CURSORS = pytest.mark.parametrize(
+    "headers",
+    [None, {"Last-Event-ID": "5"}, {"last-event-id": "5"}],
+    ids=["no-caller-cursor", "caller-cursor", "lowercase-caller-cursor"],
+)
+
+
+@_EMPTY_IDS
+@_CALLER_CURSORS
+def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id(
+    empty_id: bytes, headers: dict[str, str] | None
+):
+    reconnect = _sync_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n", empty_id], headers
+    )
+
+    assert "last-event-id" not in reconnect.headers
+
+
+@_EMPTY_IDS
+@_CALLER_CURSORS
+async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id(
+    empty_id: bytes, headers: dict[str, str] | None
+):
+    reconnect = await _async_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n", empty_id], headers
+    )
+
+    assert "last-event-id" not in reconnect.headers
+
+
+def test_sync_stream_id_replaces_a_lowercase_caller_cursor():
+    reconnect = _sync_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n"], {"last-event-id": "5"}
+    )
+
+    assert reconnect.headers.get_list("last-event-id") == ["1"]
+
+
+async def test_async_stream_id_replaces_a_lowercase_caller_cursor():
+    reconnect = await _async_reconnect(
+        [b"id: 1\nevent: values\ndata: 1\n\n"], {"last-event-id": "5"}
+    )
+
+    assert reconnect.headers.get_list("last-event-id") == ["1"]
+
+
+def test_sync_stream_keeps_the_caller_cursor_until_the_stream_sends_an_id():
+    reconnect = _sync_reconnect([b"event: values\ndata: 1\n\n"], {"Last-Event-ID": "5"})
+
+    assert reconnect.headers.get_list("last-event-id") == ["5"]
+
+
+async def test_async_stream_keeps_the_caller_cursor_until_the_stream_sends_an_id():
+    reconnect = await _async_reconnect(
+        [b"event: values\ndata: 1\n\n"], {"Last-Event-ID": "5"}
+    )
+
+    assert reconnect.headers.get_list("last-event-id") == ["5"]
 
 
 # --- _sse_to_v2_dict conversion ---