From ebc0107633aa1afaff5fb440f416ab4ce59ab112 Mon Sep 17 00:00:00 2001 From: Chester Curme Date: Fri, 10 Jul 2026 10:26:18 -0400 Subject: [PATCH] x --- .../langchain/agents/middleware/__init__.py | 2 + .../langchain/agents/middleware/tool_error.py | 201 ++++++++++++++++++ .../implementations/test_tool_error.py | 56 +++++ 3 files changed, 259 insertions(+) create mode 100644 libs/langchain_v1/langchain/agents/middleware/tool_error.py create mode 100644 libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_error.py diff --git a/libs/langchain_v1/langchain/agents/middleware/__init__.py b/libs/langchain_v1/langchain/agents/middleware/__init__.py index c49f39d85e8..f2d6163fb7b 100644 --- a/libs/langchain_v1/langchain/agents/middleware/__init__.py +++ b/libs/langchain_v1/langchain/agents/middleware/__init__.py @@ -24,6 +24,7 @@ from langchain.agents.middleware.summarization import SummarizationMiddleware, T from langchain.agents.middleware.todo import TodoListMiddleware from langchain.agents.middleware.tool_call_limit import ToolCallLimitMiddleware from langchain.agents.middleware.tool_emulator import LLMToolEmulator +from langchain.agents.middleware.tool_error import ToolErrorMiddleware from langchain.agents.middleware.tool_retry import ToolRetryMiddleware from langchain.agents.middleware.tool_selection import LLMToolSelectorMiddleware from langchain.agents.middleware.types import ( @@ -78,6 +79,7 @@ __all__ = [ "TodoListMiddleware", "ToolCallLimitMiddleware", "ToolCallRequest", + "ToolErrorMiddleware", "ToolRetryMiddleware", "TriggerClause", "after_agent", diff --git a/libs/langchain_v1/langchain/agents/middleware/tool_error.py b/libs/langchain_v1/langchain/agents/middleware/tool_error.py new file mode 100644 index 00000000000..ca2df88624d --- /dev/null +++ b/libs/langchain_v1/langchain/agents/middleware/tool_error.py @@ -0,0 +1,201 @@ +"""Tool error middleware for agents.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import TYPE_CHECKING, Any + +from langchain_core.messages import ToolMessage +from langgraph.errors import GraphBubbleUp + +from langchain.agents.middleware.types import AgentMiddleware, AgentState, ContextT, ResponseT + +if TYPE_CHECKING: + from collections.abc import Awaitable + + from langgraph.types import Command + + from langchain.agents.middleware.types import ToolCallRequest + from langchain.tools import BaseTool + +Catch = tuple[type[Exception], ...] | Callable[[Exception], bool] +"""Exceptions to catch: a tuple of exception types, or a predicate ``(exc) -> bool``.""" + +OnError = Callable[[Exception, "ToolCallRequest"], "str | list[str | dict[Any, Any]]"] +"""Formatter for a caught exception, returning `ToolMessage` content.""" + + +def _should_catch(exc: Exception, catch: Catch) -> bool: + """Return whether `exc` should be caught and returned to the model.""" + if isinstance(catch, tuple): + return isinstance(exc, catch) + return bool(catch(exc)) + + +class ToolErrorMiddleware(AgentMiddleware[AgentState[ResponseT], ContextT, ResponseT]): + """Return tool-execution exceptions to the model as error `ToolMessage`s. + + Only the exceptions named in `catch` are converted into a + `ToolMessage(status="error")`; any other exception propagates and halts the run. + Langgraph control-flow signals (interrupts, parent commands) always propagate. + + `catch` is required: there is intentionally no catch-all default, so arbitrary + internal exceptions are not serialized to the model or end user. Use `on_error` + to control (and sanitize) exactly what content the model sees. + + This middleware does not retry. For retries, compose with `ToolRetryMiddleware` + placed *inner* and configured with `on_failure="error"` so exceptions reach this + middleware. + + Guidance on what to `catch`: + + - **Catch** (return to the model): anticipated, model-actionable, non-sensitive + errors — e.g. validation errors or tool-domain errors the model can correct. + - **Do not catch** (let propagate): programming bugs, auth/permission errors, and + anything whose message may carry secrets or internal infrastructure detail. + + Examples: + !!! example "Catch a specific tool error" + + ```python + from langchain.agents import create_agent + from langchain.agents.middleware import ToolErrorMiddleware + + agent = create_agent( + model, + tools=[search_tool], + middleware=[ToolErrorMiddleware(catch=(ValueError,))], + ) + ``` + + !!! example "Custom, sanitized error message" + + ```python + def on_error(exc: Exception, request: ToolCallRequest) -> str: + name = request.tool_call["name"] + return f"`{name}` failed with invalid input. Check the arguments and retry." + + + ToolErrorMiddleware(catch=(ValueError,), on_error=on_error) + ``` + """ + + def __init__( + self, + catch: Catch, + *, + on_error: OnError | None = None, + tools: list[BaseTool | str] | None = None, + ) -> None: + """Initialize `ToolErrorMiddleware`. + + Args: + catch: Exceptions to convert into an error `ToolMessage`. Either a tuple of + exception types, or a predicate `(exc) -> bool`. Exceptions that are not + caught propagate (halting the run). + on_error: Optional formatter for the `ToolMessage` content. Receives the + exception and the tool call request (tool name, args, call id). Defaults + to a conservative prose message. May return a string or a list of + content blocks. + tools: Optional list of tools or tool names to apply handling to. Can be a + list of `BaseTool` instances or tool name strings. If `None`, applies to + all tools. + """ + super().__init__() + + self.catch = catch + self.on_error = on_error + + # Extract tool names from BaseTool instances or strings + self._tool_filter: list[str] | None + if tools is not None: + self._tool_filter = [tool.name if not isinstance(tool, str) else tool for tool in tools] + else: + self._tool_filter = None + + self.tools = [] # No additional tools registered by this middleware + + def _should_handle_tool(self, tool_name: str) -> bool: + """Check if error handling should apply to this tool.""" + if self._tool_filter is None: + return True + return tool_name in self._tool_filter + + @staticmethod + def _format_error(tool_name: str, exc: Exception) -> str: + """Default formatter for a caught exception. + + Names the exception type but omits its message, which may contain sensitive + or internal detail. Provide `on_error` to include a sanitized message. + """ + return f"Tool '{tool_name}' failed with {type(exc).__name__}." + + def wrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]], + ) -> ToolMessage | Command[Any]: + """Intercept tool execution and convert caught exceptions to error messages. + + Args: + request: Tool call request with call dict, `BaseTool`, state, and runtime. + handler: Callable to execute the tool. + + Returns: + `ToolMessage` or `Command` (the final result). + """ + tool_name = request.tool.name if request.tool else request.tool_call["name"] + + if not self._should_handle_tool(tool_name): + return handler(request) + + try: + return handler(request) + except GraphBubbleUp: + # Control-flow signals (interrupts, parent commands) must propagate. + raise + except Exception as exc: + if not _should_catch(exc, self.catch): + raise + content = ( + self.on_error(exc, request) + if self.on_error is not None + else self._format_error(tool_name, exc) + ) + return ToolMessage( + content=content, + tool_call_id=request.tool_call["id"], + name=tool_name, + status="error", + ) + + async def awrap_tool_call( + self, + request: ToolCallRequest, + handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]], + ) -> ToolMessage | Command[Any]: + """Async version of `wrap_tool_call`.""" + tool_name = request.tool.name if request.tool else request.tool_call["name"] + + if not self._should_handle_tool(tool_name): + return await handler(request) + + try: + return await handler(request) + except GraphBubbleUp: + # Control-flow signals (interrupts, parent commands) must propagate. + raise + except Exception as exc: + if not _should_catch(exc, self.catch): + raise + content = ( + self.on_error(exc, request) + if self.on_error is not None + else self._format_error(tool_name, exc) + ) + return ToolMessage( + content=content, + tool_call_id=request.tool_call["id"], + name=tool_name, + status="error", + ) diff --git a/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_error.py b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_error.py new file mode 100644 index 00000000000..89e2f26ad17 --- /dev/null +++ b/libs/langchain_v1/tests/unit_tests/agents/middleware/implementations/test_tool_error.py @@ -0,0 +1,56 @@ +"""Tests for ToolErrorMiddleware functionality.""" + +import pytest +from langchain_core.messages import HumanMessage, ToolCall, ToolMessage +from langchain_core.tools import tool + +from langchain.agents.factory import create_agent +from langchain.agents.middleware import ToolErrorMiddleware +from tests.unit_tests.agents.model import FakeToolCallingModel + + +@tool +def failing_tool(value: str) -> str: + """Tool that always fails.""" + msg = f"secret detail: {value}" + raise ValueError(msg) + + +def _model() -> FakeToolCallingModel: + return FakeToolCallingModel( + tool_calls=[ + [ToolCall(name="failing_tool", args={"value": "x"}, id="1")], + [], + ] + ) + + +def test_tool_error_caught_returns_tool_message() -> None: + """A caught exception becomes an error ToolMessage; default omits the raw message.""" + agent = create_agent( + model=_model(), + tools=[failing_tool], + middleware=[ToolErrorMiddleware(catch=(ValueError,))], + ) + + result = agent.invoke({"messages": [HumanMessage("go")]}) + + tool_messages = [m for m in result["messages"] if isinstance(m, ToolMessage)] + assert len(tool_messages) == 1 + assert tool_messages[0].status == "error" + assert tool_messages[0].name == "failing_tool" + assert "ValueError" in tool_messages[0].content + # Default formatter must not leak the raw exception message. + assert "secret detail" not in tool_messages[0].content + + +def test_tool_error_uncaught_propagates() -> None: + """An exception not listed in `catch` propagates out of the agent.""" + agent = create_agent( + model=_model(), + tools=[failing_tool], + middleware=[ToolErrorMiddleware(catch=(KeyError,))], + ) + + with pytest.raises(ValueError, match="secret detail"): + agent.invoke({"messages": [HumanMessage("go")]})