# How to Build an Event-Driven Asynchronous AI Agent

> Learn to build an event-driven asynchronous AI agent using asyncio coroutines and unified Message objects. Create responsive, non-blocking AI architectures with elegant event subscription.

- Repository: [Bojie Li/ai-agent-book](https://github.com/bojieli/ai-agent-book)
- Tags: how-to-guide
- Published: 2026-07-26

---

**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`](https://github.com/bojieli/ai-agent-book/blob/main/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`](https://github.com/bojieli/ai-agent-book/blob/main/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`](https://github.com/bojieli/ai-agent-book/blob/main/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/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/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)](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/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/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.

```python
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.

```python
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.

```python
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.

```python
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/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`](https://github.com/bojieli/ai-agent-book/blob/main/gaia_agent_server.py) file demonstrates this pattern:

```python
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`](https://github.com/bojieli/ai-agent-book/blob/main/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`](https://github.com/bojieli/ai-agent-book/blob/main/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`](https://github.com/bojieli/ai-agent-book/blob/main/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.