How to Build an Event-Driven Asynchronous AI Agent

Wrap every interaction in a unified Message object, execute tasks using asyncio coroutines with Runners.streamed_run_task(), and subscribe to specific event types via the on_event callback to create a responsive, non-blocking AI agent architecture.

The ai-agent-book repository demonstrates a production-ready pattern for building an event-driven asynchronous AI agent that handles LLM calls, tool invocations, and state changes without blocking the main thread. By combining Python's asyncio event loop with a standardized event abstraction, you can stream partial results in real-time while maintaining end-to-end observability through OpenTelemetry tracing.

Three-Layer Architecture

The implementation in bojieli/ai-agent-book follows a three-layer design that separates concerns between data representation, execution, and orchestration.

Event Abstraction Layer

At the foundation, every interaction—whether an LLM response, tool result, or user input—is encapsulated in a Message object defined in aworld/core/event/base.py. This standardization allows the system to treat all data flows uniformly, carrying payload data, headers, and distributed tracing context through a single abstraction.

The Message class includes unique identifiers and optional topics (such as Constants.PLAN), enabling selective subscription and filtering throughout the pipeline.

Async Execution Kernel

The execution engine leverages asyncio to run tasks as non-blocking coroutines. In experience_agent.py, the execute_task method submits work to Runners.streamed_run_task(task), which returns a generator yielding events asynchronously.

This pattern eliminates I/O blocking during external calls like LLM inference or HTTP tool requests. The async for ... stream_events() syntax allows the agent to process partial results as they arrive, rather than waiting for complete responses.

Event-Driven Orchestration

Higher-level components subscribe to specific event kinds and react via callbacks. The PhoneAgent class in chapter9/phone-agent/agent.py demonstrates this by accepting an on_event parameter that forwards every emitted Message to a consumer function.

For multi-agent coordination, the Swarm class supports an event_driven=True mode that forwards events from each member to a shared event bus, allowing concurrent reasoning while maintaining a coherent trace log.

Key Implementation Files

Component	Purpose	Source File
Event Model	Defines the Message class with tracing support	[aworld/core/event/base.py](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/core/event/base.py)
Async Engine	Executes tasks and streams events via coroutines	[experience_agent.py](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/experience_agent.py)
Agent Callbacks	Implements on_event pattern for UI integration	[chapter9/phone-agent/agent.py](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/phone-agent/agent.py)
Tracing	Propagates OpenTelemetry context through events	[instrumentation/eventbus.py](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/aworld/trace/instrumentation/eventbus.py)
Streaming API	Exposes events via Server-Sent Events (SSE)	[gaia_agent_server.py](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/examples/gaia/gaia_agent_server.py)

Building an Async Agent Step-by-Step

1. Define the Event Handler

Create a callback function that processes incoming events based on their kind attribute. This function receives the Message payload and typically updates UI state, logs traces, or triggers subsequent actions.

def on_event(kind: str, payload) -> None:
    if kind == "tool_call":
        print(f"Executing tool: {payload}")
    elif kind == "llm_response":
        print(f"Response chunk: {payload}")

2. Initialize the Async Runner

Use Runners.streamed_run_task() to execute tasks asynchronously. This method returns a runner object that yields events through its stream_events() generator.

from aworld.core.event.base import Message
from aworld.core.runner import Runners

class SimpleAsyncAgent:
    async def run(self, user_input: str):
        task = {"messages": [{"role": "user", "content": user_input}]}
        runner = Runners.streamed_run_task(task)
        
        async for ev in runner.stream_events():
            self.handle_event(ev)
    
    def handle_event(self, ev: Message):
        print(f"[{ev.kind}] {ev.payload}")

3. Integrate the Event Callback

Pass the on_event callback when constructing agent instances. The PhoneAgent implementation demonstrates this pattern, forwarding all internal events to the provided callback function.

from chapter9.phone_agent.agent import PhoneAgent

def ui_callback(kind: str, payload):
    # Push to websocket or update terminal UI

    print(f">>> {kind}: {payload}")

agent = PhoneAgent(on_event=ui_callback)

import asyncio
asyncio.run(agent.run("What's the weather in Tokyo?"))

4. Coordinate Multiple Agents with Swarm

For complex workflows involving multiple specialized agents, instantiate a Swarm with event_driven=True. This configuration merges event streams from all member agents into a single observable sequence.

from aworld.core.agent import Swarm

swarm = Swarm(search_agent, summarizer_agent, event_driven=True)

async def orchestrate(query):
    async for ev in swarm.run(query):
        if ev.topic == "search_result":
            print(f"Search update: {ev.payload}")

import asyncio
asyncio.run(orchestrate("Explain quantum computing"))

Reference: Implementation details appear in [test_state_manager.py](https://github.com/bojieli/ai-agent-book/blob/main/chapter8/gaia-experience/AWorld/tests/test_state_manager.py).

Streaming Events to Web Clients

To expose the event stream to front-end applications, implement a FastAPI endpoint using StreamingResponse with the text/event-stream media type. The gaia_agent_server.py file demonstrates this pattern:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

@app.get("/events")
async def sse():
    async def event_generator():
        for i in range(5):
            await asyncio.sleep(1)
            yield f"data: step {i}\n\n"
    return StreamingResponse(event_generator(), media_type="text/event-stream")

This approach enables real-time UI updates, such as displaying chat message bubbles as the LLM generates text, rather than waiting for the complete response.

Observability with Distributed Tracing

The architecture integrates OpenTelemetry by propagating trace context through Message headers. In instrumentation/eventbus.py, the system extracts tracing information from incoming events and re-attaches it to downstream calls, ensuring end-to-end visibility across tool invocations and LLM inference.

When processing events, the agent preserves the event.id and tracing headers, allowing you to correlate logs across asynchronous boundaries and identify latency bottlenecks in specific tool calls or reasoning steps.

Summary

• Standardize on Message objects defined in aworld/core/event/base.py to encapsulate all agent interactions uniformly.
• Use asyncio coroutines with Runners.streamed_run_task() to execute tasks without blocking during I/O operations.
• Implement on_event callbacks to react to specific event kinds (tool_call, llm_response) in real-time.
• Enable event_driven=True on Swarm instances to coordinate multiple agents through a shared event bus.
• Stream via SSE using FastAPI's StreamingResponse to push live updates to clients.
• Propagate tracing context through the instrumentation/eventbus.py layer to maintain observability across async boundaries.

Frequently Asked Questions

What is the difference between async agents and traditional synchronous agents?

Synchronous agents block execution while waiting for LLM API responses or tool results, limiting throughput and preventing real-time updates. Asynchronous agents use asyncio to yield control during I/O operations, allowing the event loop to process other tasks and stream partial results via stream_events() as they become available.

How do I handle errors in an event-driven async agent?

Errors should be caught within the async for loop iterating over stream_events() or within the on_event callback itself. Since the Message object includes metadata fields, you can attach error states to specific event kinds and propagate them through the same callback mechanism, allowing the UI or parent orchestrator to handle retries or display error messages without crashing the event loop.

Can I mix sync and async tool calls in the same agent?

While technically possible using asyncio.run_in_executor() for synchronous functions, the architecture in ai-agent-book recommends wrapping all external calls (tools, LLM APIs, database queries) in async def functions to maintain non-blocking behavior throughout the pipeline. Mixed patterns can lead to event loop blocking and reduced concurrency.

How does the Swarm coordinate events between multiple agents?

When instantiated with event_driven=True, the Swarm class aggregates event streams from all member agents into a single sequence. Each agent emits Message objects with distinct event.id values and optional topics, allowing the Swarm to merge these streams while preserving provenance. The parent context can then filter by ev.topic or agent ID to route events appropriately.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}