feat(langgraph): add on_interrupt hook to StateGraph.compile()

match the other bullets

libs/checkpoint-postgres/langgraph/store/postgres/base.py

@@ -151,6 +151,7 @@ class PoolConfig(TypedDict, total=False):
     """Connection pool settings for PostgreSQL connections.
 
     Controls connection lifecycle and resource utilization:
+
     - Small pools (1-5) suit low-concurrency workloads
     - Larger pools handle concurrent requests but consume more resources
     - Setting max_size prevents resource exhaustion under load
@@ -166,6 +167,7 @@ class PoolConfig(TypedDict, total=False):
     """Additional connection arguments passed to each connection in the pool.
     
     Default kwargs set automatically:
+
     - autocommit: True
     - prepare_threshold: 0
     - row_factory: dict_row
@@ -656,7 +658,8 @@ class PostgresStore(BaseStore, BasePostgresStore[_pg_internal.Conn]):
             item = store.get(("users", "123"), "prefs")
         ```
 
-        Or using the convenient from_conn_string helper:
+        Or using the convenient `from_conn_string` helper:
+
         ```python
         from langgraph.store.postgres import PostgresStore
 

libs/checkpoint-sqlite/langgraph/store/sqlite/base.py

@@ -722,7 +722,8 @@ class SqliteStore(BaseSqliteStore, BaseStore):
         item = store.get(("users", "123"), "prefs")
         ```
 
-        Or using the convenient from_conn_string helper:
+        Or using the convenient `from_conn_string` helper:
+
         ```python
         from langgraph.store.sqlite import SqliteStore
 

libs/langgraph/langgraph/graph/state.py

@@ -76,6 +76,7 @@ from langgraph.types import (
     CachePolicy,
     Checkpointer,
     Command,
+    OnInterruptHook,
     RetryPolicy,
     Send,
     ensure_valid_checkpointer,
@@ -831,6 +832,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
         interrupt_after: All | list[str] | None = None,
         debug: bool = False,
         name: str | None = None,
+        on_interrupt: OnInterruptHook | None = None,
     ) -> CompiledStateGraph[StateT, ContextT, InputT, OutputT]:
         """Compiles the `StateGraph` into a `CompiledStateGraph` object.
 
@@ -850,6 +852,10 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             interrupt_after: An optional list of node names to interrupt after.
             debug: A flag indicating whether to enable debug mode.
             name: The name to use for the compiled graph.
+            on_interrupt: An optional callback that is invoked whenever the graph
+                execution is interrupted. Called with the list of `Interrupt` objects.
+
+                May be a sync function or an async coroutine function.
 
         Returns:
             CompiledStateGraph: The compiled `StateGraph`.
@@ -910,6 +916,7 @@ class StateGraph(Generic[StateT, ContextT, InputT, OutputT]):
             store=store,
             cache=cache,
             name=name or "LangGraph",
+            on_interrupt=on_interrupt,
         )
 
         compiled.attach_node(START, None)

libs/langgraph/langgraph/pregel/_loop.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import binascii
 import concurrent.futures
+import warnings
 from collections import defaultdict, deque
 from collections.abc import Callable, Iterator, Mapping, Sequence
 from contextlib import (
@@ -12,7 +13,7 @@ from contextlib import (
     ExitStack,
 )
 from datetime import datetime, timezone
-from inspect import signature
+from inspect import iscoroutinefunction, signature
 from types import TracebackType
 from typing import (
     Any,
@@ -115,6 +116,8 @@ from langgraph.types import (
     CachePolicy,
     Command,
     Durability,
+    Interrupt,
+    OnInterruptHook,
     PregelExecutableTask,
     RetryPolicy,
     Send,
@@ -157,6 +160,7 @@ class PregelLoop:
     manager: None | AsyncParentRunManager | ParentRunManager
     interrupt_after: All | Sequence[str]
     interrupt_before: All | Sequence[str]
+    on_interrupt: OnInterruptHook | None
     durability: Durability
     retry_policy: Sequence[RetryPolicy]
     cache_policy: CachePolicy | None
@@ -226,6 +230,7 @@ class PregelLoop:
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        on_interrupt: OnInterruptHook | None = None,
     ) -> None:
         self.stream = stream
         self.config = config
@@ -242,6 +247,7 @@ class PregelLoop:
         self.stream_keys = stream_keys
         self.interrupt_after = interrupt_after
         self.interrupt_before = interrupt_before
+        self.on_interrupt = on_interrupt
         self.manager = manager
         self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {})
         self.skip_done_tasks = CONFIG_KEY_CHECKPOINT_ID not in config[CONF]
@@ -865,14 +871,36 @@ class PregelLoop:
                         [{INTERRUPT: cast(GraphInterrupt, exc_value).args[0]}]
                     ),
                 )
-            # save final output
+            # save final output first, so graph state is consistent even
+            # if the on_interrupt hook raises
             self.output = read_channels(self.channels, self.output_keys)
+            # call on_interrupt hook
+            if self.on_interrupt is not None:
+                interrupts: list[Interrupt] = (
+                    list(cast(GraphInterrupt, exc_value).args[0])
+                    if exc_value is not None and exc_value.args and exc_value.args[0]
+                    else []
+                )
+                self._call_on_interrupt(interrupts)
             # suppress interrupt
             return True
         elif exc_type is None:
             # save final output
             self.output = read_channels(self.channels, self.output_keys)
 
+    def _call_on_interrupt(self, interrupts: list[Interrupt]) -> None:
+        """Call the on_interrupt hook synchronously."""
+        if self.on_interrupt is None:
+            return
+        if iscoroutinefunction(self.on_interrupt):
+            warnings.warn(
+                "Async on_interrupt hook cannot be called from sync graph execution. "
+                "Use a sync function or run the graph with astream/ainvoke.",
+                stacklevel=2,
+            )
+            return
+        self.on_interrupt(interrupts)
+
     def _emit(
         self,
         mode: StreamMode,
@@ -985,6 +1013,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        on_interrupt: OnInterruptHook | None = None,
     ) -> None:
         super().__init__(
             input,
@@ -1006,6 +1035,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             retry_policy=retry_policy,
             cache_policy=cache_policy,
             durability=durability,
+            on_interrupt=on_interrupt,
         )
         self.stack = ExitStack()
         if checkpointer:
@@ -1161,6 +1191,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         migrate_checkpoint: Callable[[Checkpoint], None] | None = None,
         retry_policy: Sequence[RetryPolicy] = (),
         cache_policy: CachePolicy | None = None,
+        on_interrupt: OnInterruptHook | None = None,
     ) -> None:
         super().__init__(
             input,
@@ -1182,6 +1213,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             retry_policy=retry_policy,
             cache_policy=cache_policy,
             durability=durability,
+            on_interrupt=on_interrupt,
         )
         self.stack = AsyncExitStack()
         if checkpointer:
@@ -1257,6 +1289,18 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             },
         )
 
+    _deferred_on_interrupt_args: list[Interrupt] | None = None
+
+    def _call_on_interrupt(self, interrupts: list[Interrupt]) -> None:
+        """Override for async loop: defer async hooks to __aexit__."""
+        if self.on_interrupt is None:
+            return
+        if iscoroutinefunction(self.on_interrupt):
+            # Defer async hooks — they will be awaited in __aexit__
+            self._deferred_on_interrupt_args = interrupts
+        else:
+            self.on_interrupt(interrupts)
+
     # context manager
 
     async def __aenter__(self) -> Self:
@@ -1315,14 +1359,25 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         exc_value: BaseException | None,
         traceback: TracebackType | None,
     ) -> bool | None:
-        # unwind stack
+        # unwind stack (calls _suppress_interrupt synchronously)
         exit_task = asyncio.create_task(
             self.stack.__aexit__(exc_type, exc_value, traceback)
         )
         try:
-            return await exit_task
+            result = await exit_task
         except asyncio.CancelledError as e:
             # Bubble up the exit task upon cancellation to permit the API
             # consumer to await it before e.g., reusing the DB connection.
             e.args = (*e.args, exit_task)
             raise
+        # Await deferred async on_interrupt hook (set by _call_on_interrupt)
+        if (
+            self._deferred_on_interrupt_args is not None
+            and self.on_interrupt is not None
+        ):
+            interrupts = self._deferred_on_interrupt_args
+            self._deferred_on_interrupt_args = None
+            coro = self.on_interrupt(interrupts)
+            if coro is not None:
+                await coro
+        return result

libs/langgraph/langgraph/pregel/main.py

@@ -22,7 +22,6 @@ from inspect import isclass
 from typing import (
     Any,
     Generic,
-    Literal,
     cast,
     get_type_hints,
 )
@@ -39,7 +38,6 @@ from langchain_core.runnables.config import (
     get_callback_manager_for_config,
 )
 from langchain_core.runnables.graph import Graph
-from langchain_core.runnables.schema import CustomStreamEvent, StandardStreamEvent
 from langgraph.cache.base import BaseCache
 from langgraph.checkpoint.base import (
     BaseCheckpointSaver,
@@ -140,6 +138,7 @@ from langgraph.types import (
     Command,
     Durability,
     Interrupt,
+    OnInterruptHook,
     Send,
     StateSnapshot,
     StateUpdate,
@@ -624,6 +623,12 @@ class Pregel(
     context_schema: type[ContextT] | None = None
     """Specifies the schema for the context object that will be passed to the workflow."""
 
+    on_interrupt: OnInterruptHook | None = None
+    """Optional callback invoked when the graph execution is interrupted.
+
+    Called with the list of `Interrupt` objects whenever the graph pauses.
+    May be a sync or async callable."""
+
     config: RunnableConfig | None = None
 
     name: str = "LangGraph"
@@ -654,6 +659,7 @@ class Pregel(
         config: RunnableConfig | None = None,
         trigger_to_nodes: Mapping[str, Sequence[str]] | None = None,
         name: str = "LangGraph",
+        on_interrupt: OnInterruptHook | None = None,
         **deprecated_kwargs: Unpack[DeprecatedKwargs],
     ) -> None:
         if (
@@ -697,6 +703,7 @@ class Pregel(
         )
         self.cache_policy = cache_policy
         self.context_schema = context_schema
+        self.on_interrupt = on_interrupt
         self.config = config
         self.trigger_to_nodes = trigger_to_nodes or {}
         self.name = name
@@ -2601,6 +2608,7 @@ class Pregel(
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
                 cache_policy=self.cache_policy,
+                on_interrupt=self.on_interrupt,
             ) as loop:
                 # create runner
                 runner = PregelRunner(
@@ -2910,6 +2918,7 @@ class Pregel(
                 migrate_checkpoint=self._migrate_checkpoint,
                 retry_policy=self.retry_policy,
                 cache_policy=self.cache_policy,
+                on_interrupt=self.on_interrupt,
             ) as loop:
                 # create runner
                 runner = PregelRunner(
@@ -3023,61 +3032,6 @@ class Pregel(
             await asyncio.shield(run_manager.on_chain_error(e))
             raise
 
-    async def astream_events(
-        self,
-        input: InputT | Command | None,
-        config: RunnableConfig | None = None,
-        *,
-        context: ContextT | None = None,
-        version: Literal["v2"],
-        include_names: Sequence[str] | None = None,
-        include_types: Sequence[str] | None = None,
-        include_tags: Sequence[str] | None = None,
-        exclude_names: Sequence[str] | None = None,
-        exclude_types: Sequence[str] | None = None,
-        exclude_tags: Sequence[str] | None = None,
-        **kwargs: Any,
-    ) -> AsyncIterator[StandardStreamEvent | CustomStreamEvent]:
-        """Stream events from the graph execution.
-
-        This method extends the base Runnable.astream_events with support for
-        the LangGraph `context` parameter.
-
-        Args:
-            input: The input to the graph.
-            config: The configuration to use for the run.
-            context: The static context to use for the run.
-                !!! version-added "Added in version 1.0.6"
-            version: The version of the event stream schema to use ("v2").
-            include_names: Only include events from runnables with matching names.
-            include_types: Only include events from runnables with matching types.
-            include_tags: Only include events from runnables with matching tags.
-            exclude_names: Exclude events from runnables with matching names.
-            exclude_types: Exclude events from runnables with matching types.
-            exclude_tags: Exclude events from runnables with matching tags.
-            **kwargs: Additional arguments passed to the underlying stream.
-
-        Yields:
-            Events from the graph execution.
-        """
-        async with contextlib.aclosing(
-            super().astream_events(
-                input,
-                config,
-                version=version,
-                include_names=include_names,
-                include_types=include_types,
-                include_tags=include_tags,
-                exclude_names=exclude_names,
-                exclude_types=exclude_types,
-                exclude_tags=exclude_tags,
-                context=context,
-                **kwargs,
-            )
-        ) as stream:  # type: ignore[type-var]
-            async for event in stream:
-                yield event
-
     def invoke(
         self,
         input: InputT | Command | None,

libs/langgraph/langgraph/pregel/remote.py

@@ -907,7 +907,6 @@ class RemoteGraph(PregelProtocol):
         input: Any,
         config: RunnableConfig | None = None,
         *,
-        context: Any = None,
         version: Literal["v1", "v2"],
         include_names: Sequence[All] | None = None,
         include_types: Sequence[All] | None = None,

libs/langgraph/langgraph/types.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import sys
 from collections import deque
-from collections.abc import Callable, Hashable, Sequence
+from collections.abc import Awaitable, Callable, Hashable, Sequence
 from dataclasses import asdict, dataclass
 from typing import (
     TYPE_CHECKING,
@@ -56,6 +56,7 @@ __all__ = (
     "Durability",
     "interrupt",
     "Overwrite",
+    "OnInterruptHook",
     "ensure_valid_checkpointer",
 )
 
@@ -109,6 +110,18 @@ StreamWriter = Callable[[Any], None]
 Always injected into nodes if requested as a keyword argument, but it's a no-op
 when not using `stream_mode="custom"`."""
 
+OnInterruptHook = (
+    Callable[[list["Interrupt"]], None] | Callable[[list["Interrupt"]], Awaitable[None]]
+)
+"""Callback invoked when a graph execution is interrupted.
+
+Called with the list of `Interrupt` objects whenever the graph pauses due to
+an `interrupt()` call or `interrupt_before`/`interrupt_after` configuration.
+
+May be a regular function or an async coroutine function. Async hooks are
+awaited in async graph execution; in sync execution only sync hooks are called.
+"""
+
 _DC_KWARGS = {"kw_only": True, "slots": True, "frozen": True}
 
 

libs/langgraph/tests/test_large_cases.py

@@ -3038,7 +3038,7 @@ def test_message_graph(
     # add an extra message as if it came from "tools" node
     app_w_interrupt.update_state(config, ("ai", "an extra message"), as_node="tools")
 
-    # extra message is coerced BaseMessge and appended
+    # extra message is coerced BaseMessage and appended
     # now the next node is "agent" per the graph edges
     assert app_w_interrupt.get_state(config) == StateSnapshot(
         values=[
@@ -3762,7 +3762,7 @@ def test_root_graph(
     # add an extra message as if it came from "tools" node
     app_w_interrupt.update_state(config, ("ai", "an extra message"), as_node="tools")
 
-    # extra message is coerced BaseMessge and appended
+    # extra message is coerced BaseMessage and appended
     # now the next node is "agent" per the graph edges
     assert app_w_interrupt.get_state(config) == StateSnapshot(
         values=[

libs/langgraph/tests/test_pregel.py

@@ -8893,3 +8893,151 @@ def test_fork_does_not_apply_pending_writes(
 
     # Should be: 1 (input) + 20 (forked node_a) + 100 (node_b) = 121
     assert result == {"value": 121}
+
+
+def test_on_interrupt_hook_with_interrupt_call(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that on_interrupt hook fires when interrupt() is called in a node."""
+    hook_calls: list[list[Interrupt]] = []
+
+    def my_on_interrupt(interrupts: list[Interrupt]) -> None:
+        hook_calls.append(interrupts)
+
+    class State(TypedDict):
+        value: str
+
+    def ask_human(state: State) -> dict:
+        answer = interrupt("what should I do?")
+        return {"value": answer}
+
+    builder = StateGraph(State)
+    builder.add_node("ask", ask_human)
+    builder.add_edge(START, "ask")
+
+    graph = builder.compile(
+        checkpointer=sync_checkpointer,
+        on_interrupt=my_on_interrupt,
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # First invocation: should trigger interrupt and call the hook
+    result = list(graph.stream({"value": ""}, config))
+    assert len(result) == 1
+    assert "__interrupt__" in result[0]
+
+    # Hook should have been called once with the interrupt data
+    assert len(hook_calls) == 1
+    assert len(hook_calls[0]) == 1
+    assert hook_calls[0][0].value == "what should I do?"
+
+    # Resume — no new interrupt, hook should not fire again
+    hook_calls.clear()
+    result = list(graph.stream(Command(resume="do this"), config))
+    assert any("ask" in chunk for chunk in result)
+    assert len(hook_calls) == 0
+
+
+def test_on_interrupt_hook_with_interrupt_before(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that on_interrupt hook fires for interrupt_before config."""
+    hook_calls: list[list[Interrupt]] = []
+
+    def my_on_interrupt(interrupts: list[Interrupt]) -> None:
+        hook_calls.append(interrupts)
+
+    class State(TypedDict):
+        value: int
+
+    def add_one(state: State) -> dict:
+        return {"value": state["value"] + 1}
+
+    builder = StateGraph(State)
+    builder.add_node("add_one", add_one)
+    builder.add_edge(START, "add_one")
+
+    graph = builder.compile(
+        checkpointer=sync_checkpointer,
+        interrupt_before=["add_one"],
+        on_interrupt=my_on_interrupt,
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # Should interrupt before add_one runs
+    result = list(graph.stream({"value": 0}, config))
+    assert any("__interrupt__" in chunk for chunk in result)
+
+    # Hook should have been called (empty interrupt list for config-level interrupts)
+    assert len(hook_calls) == 1
+    assert hook_calls[0] == []
+
+
+def test_on_interrupt_hook_not_called_without_interrupt(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that on_interrupt hook is NOT called when graph completes normally."""
+    hook_calls: list[list[Interrupt]] = []
+
+    def my_on_interrupt(interrupts: list[Interrupt]) -> None:
+        hook_calls.append(interrupts)
+
+    class State(TypedDict):
+        value: int
+
+    def add_one(state: State) -> dict:
+        return {"value": state["value"] + 1}
+
+    builder = StateGraph(State)
+    builder.add_node("add_one", add_one)
+    builder.add_edge(START, "add_one")
+
+    graph = builder.compile(
+        checkpointer=sync_checkpointer,
+        on_interrupt=my_on_interrupt,
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    result = graph.invoke({"value": 0}, config)
+    assert result == {"value": 1}
+
+    # Hook should NOT have been called
+    assert len(hook_calls) == 0
+
+
+def test_on_interrupt_hook_exception_propagates(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that exceptions in the on_interrupt hook propagate to the caller."""
+
+    def bad_hook(interrupts: list[Interrupt]) -> None:
+        raise RuntimeError("hook exploded")
+
+    class State(TypedDict):
+        value: str
+
+    def ask(state: State) -> dict:
+        answer = interrupt("question")
+        return {"value": answer}
+
+    builder = StateGraph(State)
+    builder.add_node("ask", ask)
+    builder.add_edge(START, "ask")
+
+    graph = builder.compile(
+        checkpointer=sync_checkpointer,
+        on_interrupt=bad_hook,
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # Hook error should propagate
+    with pytest.raises(RuntimeError, match="hook exploded"):
+        list(graph.stream({"value": ""}, config))
+
+    # Graph state should still be checkpointed and resumable despite the hook error
+    result = list(graph.stream(Command(resume="answer"), config))
+    assert any("ask" in chunk for chunk in result)

libs/langgraph/tests/test_pregel_async.py

@@ -9345,3 +9345,117 @@ async def test_fork_does_not_apply_pending_writes(
 
     # 1 (input) + 20 (forked node_a) + 100 (node_b) = 121
     assert result == {"value": 121}
+
+
+async def test_on_interrupt_hook_async_with_interrupt_call(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that an async on_interrupt hook fires when interrupt() is called."""
+    hook_calls: list[list[Interrupt]] = []
+
+    async def my_on_interrupt(interrupts: list[Interrupt]) -> None:
+        hook_calls.append(interrupts)
+
+    class State(TypedDict):
+        value: str
+
+    def ask_human(state: State) -> dict:
+        answer = interrupt("what should I do?")
+        return {"value": answer}
+
+    builder = StateGraph(State)
+    builder.add_node("ask", ask_human)
+    builder.add_edge(START, "ask")
+
+    graph = builder.compile(
+        checkpointer=async_checkpointer,
+        on_interrupt=my_on_interrupt,
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # First invocation: should trigger interrupt and call the async hook
+    result = [chunk async for chunk in graph.astream({"value": ""}, config)]
+    assert len(result) == 1
+    assert "__interrupt__" in result[0]
+
+    # Hook should have been called once with the interrupt data
+    assert len(hook_calls) == 1
+    assert len(hook_calls[0]) == 1
+    assert hook_calls[0][0].value == "what should I do?"
+
+    # Resume — no new interrupt, hook should not fire again
+    hook_calls.clear()
+    result = [chunk async for chunk in graph.astream(Command(resume="do this"), config)]
+    assert any("ask" in chunk for chunk in result)
+    assert len(hook_calls) == 0
+
+
+async def test_on_interrupt_hook_sync_in_async_graph(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that a sync on_interrupt hook works in async graph execution."""
+    hook_calls: list[list[Interrupt]] = []
+
+    def my_sync_hook(interrupts: list[Interrupt]) -> None:
+        hook_calls.append(interrupts)
+
+    class State(TypedDict):
+        value: str
+
+    def ask_human(state: State) -> dict:
+        answer = interrupt("question?")
+        return {"value": answer}
+
+    builder = StateGraph(State)
+    builder.add_node("ask", ask_human)
+    builder.add_edge(START, "ask")
+
+    graph = builder.compile(
+        checkpointer=async_checkpointer,
+        on_interrupt=my_sync_hook,
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    result = [chunk async for chunk in graph.astream({"value": ""}, config)]
+    assert "__interrupt__" in result[0]
+
+    # Sync hook should work fine in async execution
+    assert len(hook_calls) == 1
+    assert hook_calls[0][0].value == "question?"
+
+
+async def test_on_interrupt_hook_async_exception_propagates(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Test that exceptions in the async on_interrupt hook propagate."""
+
+    async def bad_hook(interrupts: list[Interrupt]) -> None:
+        raise RuntimeError("async hook exploded")
+
+    class State(TypedDict):
+        value: str
+
+    def ask(state: State) -> dict:
+        answer = interrupt("question")
+        return {"value": answer}
+
+    builder = StateGraph(State)
+    builder.add_node("ask", ask)
+    builder.add_edge(START, "ask")
+
+    graph = builder.compile(
+        checkpointer=async_checkpointer,
+        on_interrupt=bad_hook,
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # Hook error should propagate
+    with pytest.raises(RuntimeError, match="async hook exploded"):
+        [chunk async for chunk in graph.astream({"value": ""}, config)]
+
+    # Graph state should still be checkpointed and resumable despite the hook error
+    result = [chunk async for chunk in graph.astream(Command(resume="answer"), config)]
+    assert any("ask" in chunk for chunk in result)

libs/langgraph/tests/test_runtime.py

@@ -389,45 +389,3 @@ def test_context_coercion_pydantic_validation_errors() -> None:
         compiled.invoke(
             {"message": "test"}, context={"api_key": "sk_test", "timeout": "not_an_int"}
         )
-
-
-@pytest.mark.anyio
-async def test_context_with_astream_events() -> None:
-    """Test that context is properly passed through astream_events."""
-
-    @dataclass
-    class Context:
-        api_key: str
-
-    class State(TypedDict):
-        message: str
-
-    def node_with_context(state: State, runtime: Runtime[Context]) -> dict[str, Any]:
-        return {"message": f"api_key: {runtime.context.api_key}"}
-
-    graph = StateGraph(state_schema=State, context_schema=Context)
-    graph.add_node("node", node_with_context)
-    graph.add_edge(START, "node")
-    graph.add_edge("node", END)
-    compiled = graph.compile()
-
-    events = []
-    async for event in compiled.astream_events(
-        {"message": "test"},
-        version="v2",
-        context=Context(api_key="sk_events_123"),
-    ):
-        events.append(event)
-
-    # Verify we got events
-    assert len(events) > 0
-
-    # Find the final on_chain_end event (no parent_ids means it's the root)
-    end_events = [
-        e for e in events if e["event"] == "on_chain_end" and not e.get("parent_ids")
-    ]
-    assert len(end_events) == 1
-
-    # Verify the output contains our context value
-    output = end_events[0]["data"]["output"]
-    assert output["message"] == "api_key: sk_events_123"

libs/prebuilt/langgraph/prebuilt/tool_node.py

@@ -15,12 +15,12 @@ The module implements design patterns for:
 
 Key Components:
 
-- `ToolNode`: Main class for executing tools in LangGraph workflows
-- `InjectedState`: Annotation for injecting graph state into tools
-- `InjectedStore`: Annotation for injecting persistent store into tools
-- `ToolRuntime`: Runtime information for tools, bundling together `state`, `context`,
+- [`ToolNode`][langgraph.prebuilt.ToolNode]: Main class for executing tools in LangGraph workflows
+- [`InjectedState`][langgraph.prebuilt.InjectedState]: Annotation for injecting graph state into tools
+- [`InjectedStore`][langgraph.prebuilt.InjectedStore]: Annotation for injecting persistent store into tools
+- [`ToolRuntime`][langgraph.prebuilt.ToolRuntime]: Runtime information for tools, bundling together `state`, `context`,
     `config`, `stream_writer`, `tool_call_id`, and `store`
-- `tools_condition`: Utility function for conditional routing based on tool calls
+- [`tools_condition`][langgraph.prebuilt.tools_condition]: Utility function for conditional routing based on tool calls
 
 Typical Usage:
     ```python
@@ -614,8 +614,16 @@ class ToolNode(RunnableCallable):
     persistent storage, and control flow. Manages parallel execution,
     error handling.
 
+    Use `ToolNode` when building custom workflows that require fine-grained control over
+    tool execution—for example, custom routing logic, specialized error handling, or
+    non-standard agent architectures.
+
+    For standard ReAct-style agents, use [`create_agent`][langchain.agents.create_agent]
+    instead. It uses `ToolNode` internally with sensible defaults for the agent loop,
+    conditional routing, and error handling.
+
     Input Formats:
-        1. Graph state with `messages` key that has a list of messages:
+        1. **Graph state** with `messages` key that has a list of messages:
             - Common representation for agentic workflows
             - Supports custom messages key via `messages_key` parameter