test_create_agent_async_jump() — langchain Function Reference

Source Code

libs/langchain_v1/tests/unit_tests/agents/middleware/core/test_framework.py lines 665–697

async def test_create_agent_async_jump() -> None:
    """Test async invoke with async middleware using jump_to."""
    calls = []

    class AsyncMiddlewareOne(AgentMiddleware):
        async def abefore_model(self, state: AgentState[Any], runtime: Runtime) -> None:
            calls.append("AsyncMiddlewareOne.abefore_model")

    class AsyncMiddlewareTwo(AgentMiddleware):
        @hook_config(can_jump_to=["end"])
        async def abefore_model(self, state: AgentState[Any], runtime: Runtime) -> dict[str, Any]:
            calls.append("AsyncMiddlewareTwo.abefore_model")
            return {"jump_to": "end"}

    @tool
    def my_tool_jump(value: str) -> str:
        """A great tool."""
        calls.append("my_tool_jump")
        return value.upper()

    agent = create_agent(
        model=FakeToolCallingModel(
            tool_calls=[[ToolCall(id="1", name="my_tool_jump", args={"value": "yo"})]],
        ),
        tools=[my_tool_jump],
        system_prompt="You are a helpful assistant.",
        middleware=[AsyncMiddlewareOne(), AsyncMiddlewareTwo()],
    )

    result = await agent.ainvoke({"messages": []})

    assert result == {"messages": []}
    assert calls == ["AsyncMiddlewareOne.abefore_model", "AsyncMiddlewareTwo.abefore_model"]