async def test_workflow_execution(calculator_agent, retriever_agent):
    """Test basic workflow execution with agent handoff."""
    workflow = AgentWorkflow(
        agents=[calculator_agent, retriever_agent],
        root_agent="retriever",
    )

    memory = ChatMemoryBuffer.from_defaults()
    handler = workflow.run(user_msg="Can you add 5 and 3?", memory=memory)

    events = []
    async for event in handler.stream_events():
        events.append(event)

    response = await handler

    # Verify we got events indicating handoff and calculation
    assert any(
        ev.current_agent_name == "retriever"
        if hasattr(ev, "current_agent_name")
        else False
        for ev in events
    )
    assert any(
        ev.current_agent_name == "calculator"
        if hasattr(ev, "current_agent_name")
        else False
        for ev in events
    )
    assert "8" in str(response.response)
Test basic workflow execution with agent handoff.
llama-index-core/tests/agent/workflow/test_multi_agent_workflow.py