> ## Documentation Index
> Fetch the complete documentation index at: https://thethirdpenco-feat-tool-lifecycle-events.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# 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()`:

```python theme={null}
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

```python theme={null}
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.

```python theme={null}
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.

<Note>
  * **`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.
</Note>

**JavaScript Client:**

```javascript theme={null}
// 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.

```python theme={null}
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:

```json theme={null}
{
  "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).

```python theme={null}
# 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
```