test_wrap_tool_call_mixed_passthrough_and_intercepting() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_wrap_tool_call.py lines 678–744

def test_wrap_tool_call_mixed_passthrough_and_intercepting() -> None:
    """Test composition with mix of pass-through and intercepting handlers."""
    call_log = []

    @wrap_tool_call(name="FirstPassthrough")
    def first_passthrough(
        request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]]
    ) -> ToolMessage | Command[Any]:
        call_log.append("first_before")
        response = handler(request)
        call_log.append("first_after")
        return response

    @wrap_tool_call(name="SecondIntercepting")
    def second_intercepting(
        request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]]
    ) -> ToolMessage | Command[Any]:
        call_log.append("second_intercept")
        # Call handler but ignore result
        _ = handler(request)
        # Return custom result
        return ToolMessage(
            content="intercepted_result",
            tool_call_id=request.tool_call["id"],
            name=request.tool_call["name"],
        )

    @wrap_tool_call(name="ThirdPassthrough")
    def third_passthrough(
        request: ToolCallRequest, handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]]
    ) -> ToolMessage | Command[Any]:
        call_log.append("third_called")
        response = handler(request)
        call_log.append("third_after")
        return response

    model = FakeToolCallingModel(
        tool_calls=[
            [ToolCall(name="search", args={"query": "test"}, id="1")],
            [],
        ]
    )

    agent = create_agent(
        model=model,
        tools=[search],
        middleware=[first_passthrough, second_intercepting, third_passthrough],
        checkpointer=InMemorySaver(),
    )

    result = agent.invoke(
        {"messages": [HumanMessage("Search")]},
        {"configurable": {"thread_id": "test"}},
    )

    # All middleware are called, second intercepts and returns custom result
    assert call_log == [
        "first_before",
        "second_intercept",
        "third_called",
        "third_after",
        "first_after",
    ]

    tool_messages = [m for m in result["messages"] if isinstance(m, ToolMessage)]
    assert len(tool_messages) == 1
    assert "intercepted_result" in tool_messages[0].content