test_mixed_middleware_composition_async_path_fails_with_sync_only() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_sync_async_wrappers.py lines 265–312

    async def test_mixed_middleware_composition_async_path_fails_with_sync_only(
        self,
    ) -> None:
        """Multiple middleware on async path fails if any are sync-only."""

        class SyncOnlyMiddleware(AgentMiddleware):
            name = "SyncOnly"

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

        class AsyncOnlyMiddleware(AgentMiddleware):
            name = "AsyncOnly"

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

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

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

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