Compare commits

..
Author SHA1 Message Date
Chester Curme f6d8e63e21 cr 2026-09-16 11:16:26 -04:00
Chester Curme a4160c8f37 fix 2026-09-16 11:06:22 -04:00
Chester Curme dcd325e581 add available_tools to ToolCallRequest 2026-09-16 10:43:28 -04:00
Chester Curme ab478cb40f nit 2026-09-15 12:04:23 -04:00
Chester Curme 218887d7ce prefer ToolCallRequest.tool_call['name'] for deciding what tool is executed in ToolNode 2026-09-15 11:33:32 -04:00
10 changed files with 177 additions and 231 deletions
+11 -66
View File
@@ -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,
),
)
)
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "langgraph"
version = "1.2.12"
version = "1.2.11"
description = "Building stateful, multi-actor applications with LLMs"
authors = []
requires-python = ">=3.10"
+1 -153
View File
@@ -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)]
}
-2
View File
@@ -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",
},
),
+4 -4
View File
@@ -1437,7 +1437,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "1.2.12"
version = "1.2.11"
source = { editable = "." }
dependencies = [
{ name = "langchain-core" },
@@ -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]]
@@ -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:
+117 -1
View File
@@ -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(),
)
+1 -1
View File
@@ -286,7 +286,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "1.2.12"
version = "1.2.11"
source = { editable = "../langgraph" }
dependencies = [
{ name = "langchain-core" },
-2
View File
@@ -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):
+1 -1
View File
@@ -299,7 +299,7 @@ wheels = [
[[package]]
name = "langgraph"
version = "1.2.12"
version = "1.2.11"
source = { editable = "../langgraph" }
dependencies = [
{ name = "langchain-core" },