MultipleDynamicToolsMiddleware Class — langchain Architecture

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_dynamic_tools.py lines 89–131

class MultipleDynamicToolsMiddleware(AgentMiddleware):
    """Middleware that dynamically adds multiple tools (sync and async)."""

    def wrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelCallResult:
        updated = request.override(tools=[*request.tools, dynamic_tool, another_dynamic_tool])
        return handler(updated)

    async def awrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
    ) -> ModelCallResult:
        updated = request.override(tools=[*request.tools, dynamic_tool, another_dynamic_tool])
        return await handler(updated)

    def _handle_tool(self, request: ToolCallRequest) -> ToolCallRequest | None:
        """Return updated request if this is a dynamic tool, else None."""
        tool_name = request.tool_call["name"]
        if tool_name == "dynamic_tool":
            return request.override(tool=dynamic_tool)
        if tool_name == "another_dynamic_tool":
            return request.override(tool=another_dynamic_tool)
        return None

    def wrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
    ) -> ToolMessage | Command[Any]:
        updated = self._handle_tool(request)
        return handler(updated or request)

    async def awrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
    ) -> ToolMessage | Command[Any]:
        updated = self._handle_tool(request)
        return await handler(updated or request)