Agent Events

Emit real-time updates with automatic replay and persistence

Agents in ai-query are designed to be real-time. Instead of just waiting for a final response, agents can emit events throughout their execution to keep users informed.

The event system is built on three core pillars:

1. Transport Agnostic - Emit once, deliver via SSE, WebSocket, or HTTP.
2. Durable - Events are persisted to storage (SQLite, Postgres, etc.).
3. Resumable - Clients can reconnect and automatically replay missed events.

The emit() API

The implementation is simple. Any method in your Agent subclass can call self.emit():

async def emit(self, event: str, data: dict[str, Any]) -> int

• event: A string identifying the event type (e.g., “status”, “log”, “chunk”)
• data: A JSON-serializable dictionary with the payload
• Returns: A unique, strictly increasing integer ID for the event

Basic Example

class SearchAgent(Agent):
    async def search(self, query: str):
        # 1. Notify user that work started
        await self.emit("status", {"text": f"Searching for: {query}"})

        # 2. Emit intermediate findings
        results = await self.perform_search(query)
        await self.emit("log", {"level": "info", "message": f"Found {len(results)} items"})

        # 3. Stream chunks if generating text
        async for chunk in self.stream_summary(results):
             await self.emit("chunk", {"content": chunk})

Event Persistence & Replay

One of the most powerful features is Automatic Event Replay.

In distributed systems, connections drop. Mobile users lose signal. If an agent emits an event while the user is disconnected, that information is usually lost.

ai-query solves this with the enable_event_log flag.

class MyAgent(Agent):
    enable_event_log = True  # <--- Enables persistence

When enabled:

1. Every emitted event is saved to the agent’s storage.
2. Events are assigned a strictly increasing ID (1, 2, 3…).
3. Clients can connect with a last_event_id parameter.
4. The server automatically sends all events with ID > last_event_id before sending new ones.

This guarantees exactly-once delivery semantics for your UI, even on flaky networks.

Consuming Events

The way you consume events depends on the transport.

Server-Sent Events (SSE)

The recommended way to consume events is via an SSE stream.

• AgentServer: Exposes a long-running SSE stream at GET /agent/{id}/events.
• Serverless/Adapters: Chat streaming is done via POST /agent/{id}/chat?stream=true. Emitted events are injected into this response stream.

JavaScript Client:

// For AgentServer
const eventSource = new EventSource("/agent/my-agent/events?last_event_id=100");

eventSource.addEventListener("status", (e) => {
  const data = JSON.parse(e.data);
  console.log("Status:", data.text);
});

// Browser EventSource handles Last-Event-ID automatically!

Python Client (connect) The connect() client handles SSE transparently for streaming.

from ai_query import connect

agent = connect("http://api.myapp.com/agent/my-agent")

async for chunk in agent.stream("Tell me a story"):
    print(chunk, end="")

WebSocket

If you connect via WebSocket (/agent/{agent_id}/ws), events are sent as JSON messages:

{
  "type": "status",
  "id": 15,
  "data": { "text": "Searching..." }
}

Custom Transports

If you use FastAPI or another framework without our adapters, you can inject a custom handler to deliver events however you like (e.g., to a message queue or a custom socket).

# Function to handle emitted events
async def my_handler(event: str, data: dict, event_id: int):
    print(f"Agent emitted {event} (ID: {event_id})")
    await websocket.send_json({...})

# Inject the handler
agent._emit_handler = my_handler