x

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",

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",
+            )

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")]})