test_dynamic_tool_chained_middleware() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_dynamic_tools.py lines 347–409

async def test_dynamic_tool_chained_middleware(*, use_async: bool) -> None:
    """Test dynamic tools work with multiple middleware in chain."""
    call_log: list[str] = []

    class LoggingMiddleware(AgentMiddleware):
        def __init__(self, label: str) -> None:
            self._label = label

        def wrap_model_call(
            self,
            request: ModelRequest,
            handler: Callable[[ModelRequest], ModelResponse],
        ) -> ModelCallResult:
            call_log.append(f"{self._label}_model")
            return handler(request)

        async def awrap_model_call(
            self,
            request: ModelRequest,
            handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
        ) -> ModelCallResult:
            call_log.append(f"{self._label}_model")
            return await handler(request)

        def wrap_tool_call(
            self,
            request: ToolCallRequest,
            handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
        ) -> ToolMessage | Command[Any]:
            call_log.append(f"{self._label}_tool")
            return handler(request)

        async def awrap_tool_call(
            self,
            request: ToolCallRequest,
            handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
        ) -> ToolMessage | Command[Any]:
            call_log.append(f"{self._label}_tool")
            return await handler(request)

    model = FakeToolCallingModel(
        tool_calls=[
            [ToolCall(name="dynamic_tool", args={"value": "chained"}, id="1")],
            [],
        ]
    )

    agent = create_agent(
        model=model,
        tools=[static_tool],
        middleware=[LoggingMiddleware("first"), DynamicToolMiddleware()],
        checkpointer=InMemorySaver(),
    )

    result = await invoke_agent(agent, "Use the dynamic tool", use_async=use_async)

    tool_messages = get_tool_messages(result)
    assert len(tool_messages) == 1
    assert tool_messages[0].name == "dynamic_tool"

    # Verify middleware chain was called
    assert "first_model" in call_log
    assert "first_tool" in call_log