cr

libs/langgraph/langgraph/types.py

@@ -12,23 +12,15 @@ from typing import (
     Generic,
     Literal,
     NamedTuple,
+    TypeVar,
     final,
-    overload,
 )
 from warnings import warn
 
 from langchain_core.messages import AnyMessage
 from langchain_core.runnables import Runnable, RunnableConfig
 from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointMetadata
-from pydantic import TypeAdapter
-from typing_extensions import (
-    NotRequired,
-    TypeAliasType,
-    TypedDict,
-    TypeVar,
-    Unpack,
-    deprecated,
-)
+from typing_extensions import NotRequired, TypeAliasType, TypedDict, Unpack, deprecated
 from xxhash import xxh3_128_hexdigest
 
 from langgraph._internal._cache import default_cache_key
@@ -44,7 +36,6 @@ from langgraph.warnings import LangGraphDeprecatedSinceV10, LangGraphDeprecatedS
 # when used in standalone type aliases.
 StateT = TypeVar("StateT")
 OutputT = TypeVar("OutputT")
-ResponseT = TypeVar("ResponseT", default=Any)
 
 if TYPE_CHECKING:
     from langgraph.pregel.protocol import PregelProtocol
@@ -581,7 +572,7 @@ _DEFAULT_INTERRUPT_ID = "placeholder-id"
 
 @final
 @dataclass(init=False, slots=True)
-class Interrupt(Generic[ResponseT]):
+class Interrupt:
     """Information about an interrupt that occurred in a node.
 
     !!! version-added "Added in version 0.2.24"
@@ -605,22 +596,13 @@ class Interrupt(Generic[ResponseT]):
     id: str
     """The ID of the interrupt. Can be used to resume the interrupt directly."""
 
-    response_schema: type[ResponseT] | dict[str, Any] | None = None
-    """Schema for the value expected when resuming this interrupt, if the graph provided one.
-
-    A surfaced interrupt carries JSON Schema (a `dict`); `type[ResponseT]` records the
-    Python type at construction so `Interrupt[Decision]` is meaningful to type checkers."""
-
     def __init__(
         self,
         value: Any,
         id: str = _DEFAULT_INTERRUPT_ID,
-        *,
-        response_schema: type[ResponseT] | dict[str, Any] | None = None,
         **deprecated_kwargs: Unpack[DeprecatedKwargs],
     ) -> None:
         self.value = value
-        self.response_schema = response_schema
 
         if (
             (ns := deprecated_kwargs.get("ns", MISSING)) is not MISSING
@@ -632,18 +614,8 @@ class Interrupt(Generic[ResponseT]):
             self.id = id
 
     @classmethod
-    def from_ns(
-        cls,
-        value: Any,
-        ns: str,
-        *,
-        response_schema: type[ResponseT] | dict[str, Any] | None = None,
-    ) -> Interrupt[ResponseT]:
-        return cls(
-            value=value,
-            id=xxh3_128_hexdigest(ns.encode()),
-            response_schema=response_schema,
-        )
+    def from_ns(cls, value: Any, ns: str) -> Interrupt:
+        return cls(value=value, id=xxh3_128_hexdigest(ns.encode()))
 
     @property
     @deprecated("`interrupt_id` is deprecated. Use `id` instead.", category=None)
@@ -876,17 +848,7 @@ class Command(Generic[N], ToolOutputMixin):
     PARENT: ClassVar[Literal["__parent__"]] = "__parent__"
 
 
-@overload
-def interrupt(value: Any, *, response_schema: type[ResponseT]) -> ResponseT: ...
-
-
-@overload
-def interrupt(value: Any, *, response_schema: dict[str, Any] | None = None) -> Any: ...
-
-
-def interrupt(
-    value: Any, *, response_schema: dict[str, Any] | type | None = None
-) -> Any:
+def interrupt(value: Any) -> Any:
     """Interrupt the graph with a resumable exception from within a node.
 
     The `interrupt` function enables human-in-the-loop workflows by pausing graph
@@ -956,7 +918,7 @@ def interrupt(
         for chunk in graph.stream({\"foo\": \"abc\"}, config):
             print(chunk)
 
-        # > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06', response_schema=None),)}
+        # > {'__interrupt__': (Interrupt(value='what is your age?', id='45fda8478b2ef754419799e10992af06'),)}
 
         command = Command(resume=\"some input from a human!!!\")
 
@@ -969,20 +931,12 @@ def interrupt(
 
     Args:
         value: The value to surface to the client when the graph is interrupted.
-        response_schema: Optional schema for the value expected on resume, surfaced
-            to clients so they can render a typed input form. Accepts a JSON Schema
-            `dict` (used as-is, resume values are not validated), or a Pydantic model
-            class, `TypedDict`, or dataclass, which are converted to JSON Schema for
-            clients and used to validate the resume value; the validated object is
-            what `interrupt` returns.
 
     Returns:
-        Any: On subsequent invocations within the same node (same task to be precise), returns the value provided during the first invocation,
-            validated against `response_schema` when one that supports validation was given.
+        Any: On subsequent invocations within the same node (same task to be precise), returns the value provided during the first invocation
 
     Raises:
         GraphInterrupt: On the first invocation within the node, halts execution and surfaces the provided value to the client.
-        pydantic.ValidationError: When a resume value does not match a Pydantic model, `TypedDict`, or dataclass `response_schema`.
     """
     from langgraph._internal._constants import (
         CONFIG_KEY_CHECKPOINT_NS,
@@ -994,36 +948,27 @@ def interrupt(
     from langgraph.errors import GraphInterrupt
 
     conf = get_config()["configurable"]
-    adapter = (
-        None
-        if response_schema is None or isinstance(response_schema, dict)
-        else TypeAdapter(response_schema)
-    )
     # track interrupt index
     scratchpad = conf[CONFIG_KEY_SCRATCHPAD]
     idx = scratchpad.interrupt_counter()
     # find previous resume values
     if scratchpad.resume:
         if idx < len(scratchpad.resume):
-            v = scratchpad.resume[idx]
-            validated = adapter.validate_python(v) if adapter else v
-            conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume[: idx + 1])])
-            return validated
+            conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
+            return scratchpad.resume[idx]
     # find current resume value
     v = scratchpad.get_null_resume(True)
     if v is not None:
         assert len(scratchpad.resume) == idx, (scratchpad.resume, idx)
-        validated = adapter.validate_python(v) if adapter else v
         scratchpad.resume.append(v)
         conf[CONFIG_KEY_SEND]([(RESUME, scratchpad.resume)])
-        return validated
+        return v
     # no resume value found
     raise GraphInterrupt(
         (
             Interrupt.from_ns(
                 value=value,
                 ns=conf[CONFIG_KEY_CHECKPOINT_NS],
-                response_schema=adapter.json_schema() if adapter else response_schema,
             ),
         )
     )

libs/langgraph/tests/test_interruption.py

@@ -1,14 +1,9 @@
-from dataclasses import dataclass
-from typing import Any
-
 import pytest
 from langgraph.checkpoint.base import BaseCheckpointSaver
-from pydantic import BaseModel, ValidationError
 from typing_extensions import TypedDict
 
 from langgraph.graph import END, START, StateGraph
-from langgraph.types import Command, Durability, Interrupt, interrupt
-from tests.any_str import AnyStr
+from langgraph.types import Durability
 
 pytestmark = pytest.mark.anyio
 
@@ -95,150 +90,3 @@ async def test_interruption_without_state_updates_async(
     assert (await graph.aget_state(thread)).next == ()
     n_checkpoints = len([c async for c in graph.aget_state_history(thread)])
     assert n_checkpoints == (5 if durability != "exit" else 3)
-
-
-class Decision(BaseModel):
-    approved: bool
-    note: str | None = None
-
-
-class DecisionDict(TypedDict):
-    approved: bool
-
-
-@dataclass
-class DecisionData:
-    approved: bool
-
-
-RAW_SCHEMA = {"type": "object", "properties": {"approved": {"type": "boolean"}}}
-
-
-@pytest.mark.parametrize(
-    ("response_schema", "expected_schema", "expected_answer"),
-    [
-        (None, None, {"approved": True, "extra": 1}),
-        (RAW_SCHEMA, RAW_SCHEMA, {"approved": True, "extra": 1}),
-        (Decision, Decision.model_json_schema(), Decision(approved=True)),
-        (
-            DecisionDict,
-            {
-                "properties": {"approved": {"title": "Approved", "type": "boolean"}},
-                "required": ["approved"],
-                "title": "DecisionDict",
-                "type": "object",
-            },
-            {"approved": True},
-        ),
-        (
-            DecisionData,
-            {
-                "properties": {"approved": {"title": "Approved", "type": "boolean"}},
-                "required": ["approved"],
-                "title": "DecisionData",
-                "type": "object",
-            },
-            DecisionData(approved=True),
-        ),
-    ],
-    ids=["none", "raw_dict", "pydantic", "typeddict", "dataclass"],
-)
-def test_interrupt_response_schema(
-    sync_checkpointer: BaseCheckpointSaver,
-    response_schema: Any,
-    expected_schema: dict[str, Any] | None,
-    expected_answer: Any,
-) -> None:
-    class State(TypedDict):
-        answer: Any
-
-    def node(state: State) -> State:
-        return {
-            "answer": interrupt(
-                {"question": "approve?"}, response_schema=response_schema
-            )
-        }
-
-    graph = (
-        StateGraph(State)
-        .add_node("node", node)
-        .add_edge(START, "node")
-        .compile(checkpointer=sync_checkpointer)
-    )
-    config = {"configurable": {"thread_id": "1"}}
-    expected = Interrupt(
-        value={"question": "approve?"}, id=AnyStr(), response_schema=expected_schema
-    )
-
-    assert list(graph.stream({"answer": None}, config)) == [
-        {"__interrupt__": (expected,)}
-    ]
-    assert graph.get_state(config).tasks[0].interrupts == (expected,)
-    assert graph.invoke(Command(resume={"approved": True, "extra": 1}), config) == {
-        "answer": expected_answer
-    }
-
-
-@pytest.mark.parametrize("resume_style", ["null", "map"])
-def test_interrupt_response_schema_rejects_invalid_resume(
-    sync_checkpointer: BaseCheckpointSaver, resume_style: str
-) -> None:
-    class State(TypedDict):
-        answer: Any
-
-    def node(state: State) -> State:
-        return {"answer": interrupt("approve?", response_schema=Decision)}
-
-    graph = (
-        StateGraph(State)
-        .add_node("node", node)
-        .add_edge(START, "node")
-        .compile(checkpointer=sync_checkpointer)
-    )
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"answer": None}, config)
-    [pending] = graph.get_state(config).tasks[0].interrupts
-
-    def resume(value: dict[str, Any]) -> Command:
-        return Command(resume=value if resume_style == "null" else {pending.id: value})
-
-    with pytest.raises(ValidationError, match="approved"):
-        graph.invoke(resume({"approved": "nope"}), config)
-
-    assert graph.invoke(resume({"approved": False}), config) == {
-        "answer": Decision(approved=False)
-    }
-
-
-@pytest.mark.parametrize("resume_style", ["null", "id_map"])
-def test_interrupt_response_schema_invalid_resume_after_earlier_interrupt(
-    sync_checkpointer: BaseCheckpointSaver, resume_style: str
-) -> None:
-    class State(TypedDict):
-        answer: Any
-
-    def node(state: State) -> State:
-        first = interrupt("first")
-        second = interrupt("approve?", response_schema=Decision)
-        return {"answer": [first, second]}
-
-    graph = (
-        StateGraph(State)
-        .add_node("node", node)
-        .add_edge(START, "node")
-        .compile(checkpointer=sync_checkpointer)
-    )
-    config = {"configurable": {"thread_id": "1"}}
-    graph.invoke({"answer": None}, config)
-    graph.invoke(Command(resume="ok"), config)
-    [pending] = graph.get_state(config).tasks[0].interrupts
-
-    def resume(value: dict[str, Any]) -> Command:
-        return Command(resume=value if resume_style == "null" else {pending.id: value})
-
-    with pytest.raises(ValidationError, match="approved"):
-        graph.invoke(resume({"approved": "nope"}), config)
-
-    assert graph.invoke(resume({"approved": True}), config) == {
-        "answer": ["ok", Decision(approved=True)]
-    }

libs/langgraph/tests/test_pregel.py

@@ -5583,7 +5583,6 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
                 "interrupts": [
                     {
                         "id": AnyStr(),
-                        "response_schema": None,
                         "value": "test",
                     },
                 ],
@@ -5628,7 +5627,6 @@ def test_falsy_return_from_task(sync_checkpointer: BaseCheckpointSaver):
                         "interrupts": (
                             {
                                 "id": AnyStr(),
-                                "response_schema": None,
                                 "value": "test",
                             },
                         ),

libs/langgraph/uv.lock

@@ -3404,11 +3404,11 @@ wheels = [
 
 [[package]]
 name = "soupsieve"
-version = "2.9"
+version = "2.8.4"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/80/f1/93422647dd7e461f23d254e6b2bfa687a85b53aeb4903fcdbb74474d4584/soupsieve-2.9.tar.gz", hash = "sha256:acee8417325c5653e1377dc31eccad59eb82cbc65942afe6174c53b3aaad63fc", size = 122122, upload-time = "2026-07-19T01:35:18.425Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/47/2c/0a5f6f8ee0d5589e48c7640213ed5175d52cf540a06725b628cc1a45d6ce/soupsieve-2.8.4.tar.gz", hash = "sha256:e121fd02e975c695e4e9e8774a5ee35d74714b59307868dcc5319ad2d9e3328e", size = 121110, upload-time = "2026-05-24T13:55:57.154Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7b/d6/3185ab5ad1280319b31986898f3206dd7227cd75e293d4dba2a5e6bf27a0/soupsieve-2.9-py3-none-any.whl", hash = "sha256:a2b2c76d67df2382d245409fd71e321a571717e58463efa32ace87dcadac2c12", size = 37387, upload-time = "2026-07-19T01:35:17.106Z" },
+    { url = "https://files.pythonhosted.org/packages/5e/f5/0c41cb68dcae6b7de4fac4188a3a9589e21fb31df21ea3a2e888db95e6c9/soupsieve-2.8.4-py3-none-any.whl", hash = "sha256:e7e6b0769c8f51ed59acab6e994b00621096cfb1c640a7509295987388fbaf65", size = 37304, upload-time = "2026-05-24T13:55:55.406Z" },
 ]
 
 [[package]]

libs/prebuilt/langgraph/prebuilt/tool_node.py

@@ -135,18 +135,25 @@ class ToolCallRequest:
 
     Attributes:
         tool_call: Tool call dict with name, args, and id from model output.
+
+            If an interceptor edits `tool_call["name"]` so it differs from `tool`,
+            `tool_call["name"]` is authoritative for what tool is executed.
         tool: BaseTool instance to be invoked, or None if tool is not
             registered with the `ToolNode`. When tool is `None`, interceptors can
             handle the request without validation. If the interceptor calls `execute()`,
             validation will occur and raise an error for unregistered tools.
         state: Agent state (`dict`, `list`, or `BaseModel`).
         runtime: LangGraph runtime context (optional, `None` if outside graph).
+        available_tools: Client-side tools registered with the `ToolNode`. Provider
+            and built-in tools are not included. Use this to resolve a replacement
+            tool when redirecting a call, and set `tool` to the resolved instance.
     """
 
     tool_call: ToolCall
     tool: BaseTool | None
     state: Any
     runtime: ToolRuntime
+    available_tools: list[BaseTool] = field(default_factory=list)
 
     def __setattr__(self, name: str, value: Any) -> None:
         """Raise deprecation warning when setting attributes directly.
@@ -336,6 +343,27 @@ def msg_content_output(output: Any) -> str | list[dict]:
         return str(output)
 
 
+class ToolCallRequestMismatchError(ValueError):
+    """`tool_call["name"]` and `tool` disagree on a `ToolCallRequest`."""
+
+
+def _check_not_redirected_without_tool(
+    request: ToolCallRequest, original_name: str, original_tool: BaseTool | None
+) -> None:
+    """Raise if an interceptor renamed the call but left `tool` as the resolved one."""
+    if (
+        original_tool is not None
+        and request.tool is original_tool
+        and request.tool_call["name"] != original_name
+    ):
+        msg = (
+            f"Interceptor set tool_call name to {request.tool_call['name']!r} but left "
+            f"`tool` as {original_tool.name!r}. Redirecting a call requires setting both; "
+            f"resolve the replacement from `ToolCallRequest.available_tools`."
+        )
+        raise ToolCallRequestMismatchError(msg)
+
+
 class ToolInvocationError(ToolException):
     """An error occurred while invoking a tool due to invalid arguments.
 
@@ -1037,6 +1065,7 @@ class ToolNode(RunnableCallable):
             tool=tool,
             state=tool_runtime.state,
             runtime=tool_runtime,
+            available_tools=list(self.tools_by_name.values()),
         )
 
         config = tool_runtime.config
@@ -1046,13 +1075,18 @@ class ToolNode(RunnableCallable):
             return self._execute_tool_sync(tool_request, input_type, config)
 
         # Define execute callable that can be called multiple times
+        original_name, original_tool = call["name"], tool
+
         def execute(req: ToolCallRequest) -> ToolMessage | Command:
             """Execute tool with given request. Can be called multiple times."""
+            _check_not_redirected_without_tool(req, original_name, original_tool)
             return self._execute_tool_sync(req, input_type, config)
 
         # Call wrapper with request and execute callable
         try:
             return self._wrap_tool_call(tool_request, execute)
+        except ToolCallRequestMismatchError:
+            raise
         except Exception as e:
             # Wrapper threw an exception
             if not self._handle_tool_errors:
@@ -1184,6 +1218,7 @@ class ToolNode(RunnableCallable):
             tool=tool,
             state=tool_runtime.state,
             runtime=tool_runtime,
+            available_tools=list(self.tools_by_name.values()),
         )
 
         config = tool_runtime.config
@@ -1193,12 +1228,16 @@ class ToolNode(RunnableCallable):
             return await self._execute_tool_async(tool_request, input_type, config)
 
         # Define async execute callable that can be called multiple times
+        original_name, original_tool = call["name"], tool
+
         async def execute(req: ToolCallRequest) -> ToolMessage | Command:
             """Execute tool with given request. Can be called multiple times."""
+            _check_not_redirected_without_tool(req, original_name, original_tool)
             return await self._execute_tool_async(req, input_type, config)
 
         def _sync_execute(req: ToolCallRequest) -> ToolMessage | Command:
             """Sync execute fallback for sync wrapper."""
+            _check_not_redirected_without_tool(req, original_name, original_tool)
             return self._execute_tool_sync(req, input_type, config)
 
         # Call wrapper with request and execute callable
@@ -1208,6 +1247,8 @@ class ToolNode(RunnableCallable):
             # None check was performed above already
             self._wrap_tool_call = cast("ToolCallWrapper", self._wrap_tool_call)
             return self._wrap_tool_call(tool_request, _sync_execute)
+        except ToolCallRequestMismatchError:
+            raise
         except Exception as e:
             # Wrapper threw an exception
             if not self._handle_tool_errors:

libs/prebuilt/tests/test_on_tool_call.py

@@ -1,7 +1,7 @@
 """Unit tests for tool call interceptor in ToolNode."""
 
 import functools
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from unittest.mock import Mock
 
 import pytest
@@ -13,6 +13,7 @@ from langgraph.types import Command
 
 from langgraph.prebuilt.tool_node import (
     ToolCallRequest,
+    ToolCallRequestMismatchError,
     ToolNode,
 )
 
@@ -1471,3 +1472,118 @@ def test_tool_call_request_is_frozen() -> None:
     assert fresh_new_request.tool == add  # Other fields should remain the same
     assert fresh_new_request.state == state
     assert fresh_new_request.runtime is None
+
+
+async def test_interceptor_can_redirect_to_another_tool() -> None:
+    """Redirecting requires setting both `tool_call` and `tool`; routing follows them."""
+
+    @tool
+    def subtract(a: int, b: int) -> int:
+        """Subtract two numbers."""
+        return a - b
+
+    async def redirect(
+        request: ToolCallRequest,
+        execute: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
+    ) -> ToolMessage | Command:
+        target = next(t for t in request.available_tools if t.name == "subtract")
+        return await execute(
+            request.override(
+                tool_call={**request.tool_call, "name": "subtract"}, tool=target
+            )
+        )
+
+    node = ToolNode([add, subtract], awrap_tool_call=redirect)
+    result = await node.ainvoke(
+        [
+            AIMessage(
+                "",
+                tool_calls=[
+                    {
+                        "name": "add",
+                        "args": {"a": 5, "b": 3},
+                        "id": "1",
+                        "type": "tool_call",
+                    }
+                ],
+            )
+        ],
+        _create_config_with_runtime(),
+    )
+
+    # `add` would return 8; `subtract` returns 2.
+    assert result[0].content == "2"
+
+
+def test_interceptor_tool_call_name_and_tool_must_agree() -> None:
+    """Renaming `tool_call` without `tool` raises rather than running the wrong tool."""
+
+    def rename_only(
+        request: ToolCallRequest,
+        execute: Callable[[ToolCallRequest], ToolMessage | Command],
+    ) -> ToolMessage | Command:
+        return execute(
+            request.override(tool_call={**request.tool_call, "name": "other"})
+        )
+
+    @tool
+    def other(a: int, b: int) -> int:
+        """Another tool."""
+        return 0
+
+    # handle_tool_errors is on by default; the mismatch must not become a ToolMessage
+    node = ToolNode([add, other], wrap_tool_call=rename_only)
+    with pytest.raises(ToolCallRequestMismatchError, match="other"):
+        node.invoke(
+            [
+                AIMessage(
+                    "",
+                    tool_calls=[
+                        {
+                            "name": "add",
+                            "args": {"a": 1, "b": 2},
+                            "id": "1",
+                            "type": "tool_call",
+                        }
+                    ],
+                )
+            ],
+            _create_config_with_runtime(),
+        )
+
+
+async def test_sync_interceptor_under_ainvoke_also_validates_redirect() -> None:
+    """The sync-wrapper fallback used by `ainvoke` must validate too, not just `invoke`."""
+
+    def rename_only(
+        request: ToolCallRequest,
+        execute: Callable[[ToolCallRequest], ToolMessage | Command],
+    ) -> ToolMessage | Command:
+        return execute(
+            request.override(tool_call={**request.tool_call, "name": "other"})
+        )
+
+    @tool
+    def other(a: int, b: int) -> int:
+        """Another tool."""
+        return 0
+
+    # Only a sync wrapper is configured, so `ainvoke` routes through `_sync_execute`.
+    node = ToolNode([add, other], wrap_tool_call=rename_only)
+    with pytest.raises(ToolCallRequestMismatchError, match="other"):
+        await node.ainvoke(
+            [
+                AIMessage(
+                    "",
+                    tool_calls=[
+                        {
+                            "name": "add",
+                            "args": {"a": 1, "b": 2},
+                            "id": "1",
+                            "type": "tool_call",
+                        }
+                    ],
+                )
+            ],
+            _create_config_with_runtime(),
+        )

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

@@ -173,18 +173,8 @@ class RunModule:
         config: dict[str, Any] | None = None,
         metadata: dict[str, Any] | None = None,
         langsmith_tracing: LangSmithTracing | None = None,
-        context: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
-        """Send `run.start` to the server. Returns the result (`{"run_id": ...}`).
-
-        Args:
-            input: the run input; omitted from the wire payload when None.
-            config: the run config; omitted when None.
-            metadata: run metadata; omitted when None.
-            langsmith_tracing: tracing options; omitted when None.
-            context: per-run static context; omitted from the wire payload
-                when None (server applies its default context behavior).
-        """
+        """Send `run.start` to the server. Returns the result (`{"run_id": ...}`)."""
         params: dict[str, Any] = {"assistant_id": self._owner.assistant_id}
         if input is not None:
             params["input"] = input
@@ -194,8 +184,6 @@ class RunModule:
             params["metadata"] = metadata
         if langsmith_tracing is not None:
             params["langsmith_tracer"] = langsmith_tracing
-        if context is not None:
-            params["context"] = context
         loop = asyncio.get_running_loop()
         gate: asyncio.Future[None] = loop.create_future()
         self._owner._run_start_ready = gate
@@ -228,7 +216,6 @@ class RunModule:
         response: Any,
         *,
         interrupt_id: str | None = None,
-        context: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Reply to a server-side interrupt and resume the run.
 
@@ -237,8 +224,6 @@ class RunModule:
                 wire (protocol field name).
             interrupt_id: optional explicit id. When omitted, requires exactly
                 one outstanding interrupt and uses its id.
-            context: optional per-run static context for the resumed run;
-                forwarded with the `input.respond` command when non-None.
 
         Raises:
             RuntimeError: no outstanding interrupts; `interrupt_id` is None but
@@ -281,8 +266,6 @@ class RunModule:
                 "namespace": match["namespace"],
                 "response": response,
             }
-            if context is not None:
-                params["context"] = context
             return await self._owner._send_command("input.respond", params)
 
 

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

@@ -216,18 +216,8 @@ class SyncRunModule:
         config: dict[str, Any] | None = None,
         metadata: dict[str, Any] | None = None,
         langsmith_tracing: LangSmithTracing | None = None,
-        context: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
-        """Send `run.start` to the server. Returns the result (`{"run_id": ...}`).
-
-        Args:
-            input: the run input; omitted from the wire payload when None.
-            config: the run config; omitted when None.
-            metadata: run metadata; omitted when None.
-            langsmith_tracing: tracing options; omitted when None.
-            context: per-run static context; omitted from the wire payload
-                when None (server applies its default context behavior).
-        """
+        """Send `run.start` to the server. Returns the result (`{"run_id": ...}`)."""
         params: dict[str, Any] = {"assistant_id": self._owner.assistant_id}
         if input is not None:
             params["input"] = input
@@ -237,8 +227,6 @@ class SyncRunModule:
             params["metadata"] = metadata
         if langsmith_tracing is not None:
             params["langsmith_tracer"] = langsmith_tracing
-        if context is not None:
-            params["context"] = context
         result = self._owner._send_command("run.start", params)
         self._owner._run_seen = True
         controller = self._owner._controller
@@ -251,7 +239,6 @@ class SyncRunModule:
         response: Any,
         *,
         interrupt_id: str | None = None,
-        context: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Reply to a server-side interrupt and resume the run.
 
@@ -259,8 +246,6 @@ class SyncRunModule:
             response: the response value forwarded as `params.response` on the wire.
             interrupt_id: optional explicit id. When omitted, requires exactly one
                 outstanding interrupt.
-            context: optional per-run static context for the resumed run;
-                forwarded with the `input.respond` command when non-None.
 
         Raises:
             RuntimeError: no outstanding interrupts; `interrupt_id` is None but
@@ -297,8 +282,6 @@ class SyncRunModule:
             "namespace": match["namespace"],
             "response": response,
         }
-        if context is not None:
-            params["context"] = context
         return self._owner._send_command("input.respond", params)
 
 

libs/sdk-py/langgraph_sdk/schema.py

@@ -295,8 +295,6 @@ class Interrupt(TypedDict):
     """The value associated with the interrupt."""
     id: str
     """The ID of the interrupt. Can be used to resume the interrupt."""
-    response_schema: NotRequired[dict[str, Any]]
-    """JSON Schema for the value expected when resuming this interrupt, if the graph provided one."""
 
 
 class Thread(TypedDict):

libs/sdk-py/tests/streaming/test_sync_thread_stream.py

@@ -439,64 +439,6 @@ def test_sync_run_start_sends_command():
     }
 
 
-def test_sync_run_start_forwards_context():
-    fake = SyncFakeServer()
-    fake.script([lifecycle_completed_event(seq=1)])
-    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
-        threads = SyncThreadsClient(SyncHttpClient(raw))
-        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            thread.run.start(input={"x": 1}, context={"user_id": "u-1"})
-
-    assert fake.received_commands[0]["params"]["context"] == {"user_id": "u-1"}
-
-
-def test_sync_run_start_omits_context_when_not_provided():
-    fake = SyncFakeServer()
-    fake.script([lifecycle_completed_event(seq=1)])
-    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
-        threads = SyncThreadsClient(SyncHttpClient(raw))
-        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            thread.run.start(input={"x": 1})
-
-    assert "context" not in fake.received_commands[0]["params"]
-
-
-def test_sync_run_respond_forwards_context():
-    fake = SyncFakeServer()
-    fake.script([lifecycle_completed_event(seq=1)])
-    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
-        threads = SyncThreadsClient(SyncHttpClient(raw))
-        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            thread.run.start(input={})
-            thread.interrupts.append(
-                {"interrupt_id": "i-1", "value": None, "namespace": []}
-            )
-            thread.interrupted = True
-            thread.run.respond("yes", context={"user_id": "u-1"})
-
-    command = fake.received_commands[-1]
-    assert command["method"] == "input.respond"
-    assert command["params"]["context"] == {"user_id": "u-1"}
-
-
-def test_sync_run_respond_omits_context_when_not_provided():
-    fake = SyncFakeServer()
-    fake.script([lifecycle_completed_event(seq=1)])
-    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
-        threads = SyncThreadsClient(SyncHttpClient(raw))
-        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            thread.run.start(input={})
-            thread.interrupts.append(
-                {"interrupt_id": "i-1", "value": None, "namespace": []}
-            )
-            thread.interrupted = True
-            thread.run.respond("yes")
-
-    command = fake.received_commands[-1]
-    assert command["method"] == "input.respond"
-    assert "context" not in command["params"]
-
-
 def test_sync_events_iterates_raw_events():
 
     fake = SyncFakeServer()

libs/sdk-py/tests/streaming/test_thread_stream.py

@@ -311,28 +311,6 @@ async def test_run_start_forwards_config_metadata_and_langsmith_tracing():
     }
 
 
-async def test_run_start_forwards_context():
-    fake = FakeServer()
-    transport = httpx.ASGITransport(app=fake.app)
-    async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
-        threads = ThreadsClient(HttpClient(raw))
-        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            await thread.run.start(input={"x": 1}, context={"user_id": "u-1"})
-    params = fake.received_commands[0]["params"]
-    assert params["context"] == {"user_id": "u-1"}
-
-
-async def test_run_start_omits_context_when_not_provided():
-    fake = FakeServer()
-    transport = httpx.ASGITransport(app=fake.app)
-    async with httpx.AsyncClient(transport=transport, base_url="http://test") as raw:
-        threads = ThreadsClient(HttpClient(raw))
-        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            await thread.run.start(input={"x": 1})
-    params = fake.received_commands[0]["params"]
-    assert "context" not in params
-
-
 async def test_run_start_raises_outside_context_manager():
 
     async with httpx.AsyncClient(base_url="http://test") as raw:
@@ -638,38 +616,6 @@ async def test_run_respond_dispatches_input_respond_command():
     assert command["params"]["namespace"] == []
 
 
-async def test_run_respond_forwards_context():
-    fake = FakeServer()
-    asgi = httpx.ASGITransport(app=fake.app)
-    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
-        threads = ThreadsClient(HttpClient(raw))
-        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            await thread.run.start(input={})
-            thread.interrupts.append(
-                {"interrupt_id": "i-1", "value": None, "namespace": []}
-            )
-            thread.interrupted = True
-            await thread.run.respond("yes", context={"user_id": "u-1"})
-    params = fake.received_commands[-1]["params"]
-    assert params["context"] == {"user_id": "u-1"}
-
-
-async def test_run_respond_omits_context_when_not_provided():
-    fake = FakeServer()
-    asgi = httpx.ASGITransport(app=fake.app)
-    async with httpx.AsyncClient(transport=asgi, base_url="http://test") as raw:
-        threads = ThreadsClient(HttpClient(raw))
-        async with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            await thread.run.start(input={})
-            thread.interrupts.append(
-                {"interrupt_id": "i-1", "value": None, "namespace": []}
-            )
-            thread.interrupted = True
-            await thread.run.respond("yes")
-    params = fake.received_commands[-1]["params"]
-    assert "context" not in params
-
-
 async def test_run_respond_with_explicit_interrupt_id():
     fake = FakeServer()
     asgi = httpx.ASGITransport(app=fake.app)