TestSyncAsyncMiddlewareComposition Class — langchain Architecture

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_sync_async_wrappers.py lines 35–490

class TestSyncAsyncMiddlewareComposition:
    """Test sync/async middleware composition behavior."""

    def test_sync_only_middleware_works_on_sync_path(self) -> None:
        """Middleware with only sync wrap_tool_call works on sync path."""
        call_log = []

        class SyncOnlyMiddleware(AgentMiddleware):
            def wrap_tool_call(
                self,
                request: ToolCallRequest,
                handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
            ) -> ToolMessage | Command[Any]:
                call_log.append("sync_called")
                return handler(request)

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

        agent = create_agent(
            model=model,
            tools=[search],
            middleware=[SyncOnlyMiddleware()],
            checkpointer=InMemorySaver(),
        )

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

        assert "sync_called" in call_log
        tool_messages = [m for m in result["messages"] if isinstance(m, ToolMessage)]
        assert len(tool_messages) == 1
        assert "Results for: test" in tool_messages[0].content

    async def test_sync_only_middleware_raises_on_async_path(self) -> None:
        """Middleware with only sync wrap_tool_call raises NotImplementedError on async path."""

        class SyncOnlyMiddleware(AgentMiddleware):
            def wrap_tool_call(
                self,
                request: ToolCallRequest,
                handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
            ) -> ToolMessage | Command[Any]:
                return handler(request)

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

        agent = create_agent(
            model=model,
            tools=[search],
            middleware=[SyncOnlyMiddleware()],
            checkpointer=InMemorySaver(),
        )

        # Should raise NotImplementedError because SyncOnlyMiddleware doesn't support async path
        with pytest.raises(NotImplementedError):
            await agent.ainvoke(
                {"messages": [HumanMessage("Search")]},
                {"configurable": {"thread_id": "test"}},
            )

    async def test_async_only_middleware_works_on_async_path(self) -> None:
        """Middleware with only async awrap_tool_call works on async path."""
        call_log = []

        class AsyncOnlyMiddleware(AgentMiddleware):
            async def awrap_tool_call(
                self,
                request: ToolCallRequest,
                handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],