Building Event-Driven AI Agents: Architecture for Reactive Agent Systems

Why Event-Driven Architecture for AI Agents

Traditional request-response AI agents wait for a user to ask a question. Event-driven AI agents flip this model entirely. They sit on a message bus, listening for events — a new file uploaded, a payment processed, a sensor reading out of range — and react autonomously without human initiation.

This architecture unlocks a category of agent behavior that is impossible with synchronous designs: agents that monitor, respond, and adapt to streams of real-world activity in real time. Production systems at companies like Stripe, GitHub, and Datadog all rely on event-driven patterns to power their automation layers.

In this guide, you will build a complete event-driven agent framework using FastAPI, an in-process event bus, and async handlers that scale horizontally.

Core Concepts

An event-driven agent system has four primary components:

• Event producers — services or webhooks that emit structured events
• Event bus — the routing layer that delivers events to interested handlers
• Event handlers — functions that process specific event types
• Agent logic — the AI reasoning layer that decides what action to take

The separation between the bus and the handlers is what makes the system scalable. You can add new event types and handlers without modifying existing code.

Building the Event Bus

Start with a lightweight in-process event bus. For production systems, you would swap this for Redis Streams, RabbitMQ, or Kafka, but the handler interface stays the same.

import asyncio
from typing import Callable, Any
from dataclasses import dataclass, field
from datetime import datetime
import uuid


@dataclass
class Event:
    event_type: str
    payload: dict[str, Any]
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())


class EventBus:
    def __init__(self):
        self._handlers: dict[str, list[Callable]] = {}
        self._queue: asyncio.Queue[Event] = asyncio.Queue()

    def subscribe(self, event_type: str, handler: Callable):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    async def publish(self, event: Event):
        await self._queue.put(event)

    async def start_processing(self):
        while True:
            event = await self._queue.get()
            handlers = self._handlers.get(event.event_type, [])
            tasks = [handler(event) for handler in handlers]
            if tasks:
                await asyncio.gather(*tasks, return_exceptions=True)
            self._queue.task_done()

The EventBus class uses an asyncio queue internally. Producers call publish(), and the processing loop fans out each event to all subscribed handlers concurrently.

Registering Agent Handlers

Now wire up agent handlers that contain AI logic. Each handler subscribes to a specific event type and decides what to do based on the payload.

from openai import AsyncOpenAI

client = AsyncOpenAI()
bus = EventBus()


async def handle_support_ticket(event: Event):
    ticket = event.payload
    prompt = f"Classify this support ticket and suggest a response:\n{ticket['body']}"

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    classification = response.choices[0].message.content
    print(f"Ticket {ticket['id']} classified: {classification}")


async def handle_deployment(event: Event):
    deploy = event.payload
    if deploy["status"] == "failed":
        prompt = f"Analyze this deployment failure and suggest fixes:\n{deploy['logs']}"
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}],
        )
        print(f"Deployment fix suggestion: {response.choices[0].message.content}")


bus.subscribe("support.ticket.created", handle_support_ticket)
bus.subscribe("deployment.completed", handle_deployment)

Integrating with FastAPI

Expose the event bus through a FastAPI application so external services can push events via HTTP.

from fastapi import FastAPI
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(bus.start_processing())
    yield
    task.cancel()


app = FastAPI(lifespan=lifespan)


@app.post("/events")
async def receive_event(event_type: str, payload: dict):
    event = Event(event_type=event_type, payload=payload)
    await bus.publish(event)
    return {"event_id": event.event_id, "status": "accepted"}

The lifespan context manager starts the event processing loop when the server boots and cancels it on shutdown. Events are accepted immediately and processed asynchronously, so the HTTP response returns fast regardless of how long the AI handler takes.

Scaling Considerations

For production workloads, replace the in-process queue with a distributed message broker. Redis Streams is a good starting point because it supports consumer groups, which let you run multiple agent workers processing events in parallel without duplicating work.

The handler interface remains identical — only the bus implementation changes. This is the key architectural advantage of event-driven design: your AI logic is decoupled from your delivery infrastructure.

FAQ

When should I use event-driven agents instead of a simple API?

Use event-driven agents when you need to react to things that happen outside your control — third-party webhooks, database changes, infrastructure alerts. If the agent only responds to direct user requests, a standard API is simpler and sufficient.

How do I prevent duplicate event processing?

Store processed event IDs in a database or Redis set. Before handling an event, check if its ID has already been processed. This idempotency check is critical when using at-least-once delivery brokers like Kafka or RabbitMQ.

What happens if an agent handler fails mid-processing?

With the asyncio-based bus shown above, exceptions are caught by return_exceptions=True in asyncio.gather. For production systems, implement a dead letter queue that captures failed events with their error context so you can replay them after fixing the handler.

---

#EventDrivenArchitecture #AIAgents #FastAPI #AsyncProcessing #MessageBus #AgenticAI #LearnAI #AIEngineering