---
title: "Agent Events (SSE)"
description: "Real-time agent event streaming via Postgres LISTEN/NOTIFY and Server-Sent Events"
canonical: "https://docs.ainative.studio/docs/agent-cloud/events"
last-updated: "2026-10-13T21:42:35.442Z"
---

# Agent Events (SSE)

Source: https://docs.ainative.studio/docs/agent-cloud/events

> Real-time agent event streaming via Postgres LISTEN/NOTIFY and Server-Sent Events

Real-time push-based agent event streaming. Replaces polling-based status updates with Postgres LISTEN/NOTIFY piped through Server-Sent Events (SSE). Refs #5024

---

## Overview

Agent events flow through Postgres NOTIFY channels and are exposed as SSE streams. Each agent gets its own channel. A global channel aggregates all agent events for admin dashboards.

```
Service calls notify()
    |
    v
Postgres NOTIFY on agent_events_{agent_id}
    |
    v
Dedicated asyncpg LISTEN connection (not from pool)
    |
    v
SSE endpoint streams events to client
```

Key design decisions:
- **Dedicated connection**: LISTEN uses a standalone asyncpg connection, not the SQLAlchemy pool. Long-lived SSE connections don't consume pool slots.
- **Auto-truncation**: Payloads exceeding 7.5 KB are replaced with a slim notification containing a `fetch_url` for the full payload.
- **30-second keepalive**: Clients receive keepalive pings every 30 seconds to prevent connection timeouts.

---

## Endpoints

| Method | Path | Description |
|--------|------|-------------|
| `GET` | `/api/v1/cloud/agents/{agent_id}/events/stream` | SSE stream for a specific agent |
| `GET` | `/api/v1/cloud/agents/events/stream` | SSE stream for all agents (admin) |
| `POST` | `/api/v1/cloud/agents/{agent_id}/events/emit` | Manually emit an event (testing) |

---

## Consuming Events

### Per-Agent Stream

```javascript
const source = new EventSource(
  'https://api.ainative.studio/api/v1/cloud/agents/agent-abc-123/events/stream',
  { headers: { 'Authorization': 'Bearer <token>' } }
);

source.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data.type, data);
};
```

### Admin Stream (All Agents)

```javascript
const source = new EventSource(
  'https://api.ainative.studio/api/v1/cloud/agents/events/stream'
);

source.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(`[${data.agent_id}]`, data.type, data);
};
```

---

## Event Format

### Standard Event

```json
{
  "agent_id": "agent-abc-123",
  "type": "status_changed",
  "timestamp": "2026-07-13T10:30:00+00:00",
  "data": {
    "status": "RUNNING",
    "task_id": "task-456"
  }
}
```

### Connected Event (first event on stream open)

```json
{
  "type": "connected",
  "agent_id": "agent-abc-123",
  "channel": "agent_events_agent_abc_123",
  "timestamp": "2026-07-13T10:30:00+00:00"
}
```

### Keepalive Ping

```json
{
  "type": "keepalive",
  "timestamp": "2026-07-13T10:30:30+00:00"
}
```

### Truncated Event (payload > 7.5 KB)

When a payload exceeds the Postgres NOTIFY size limit, the event is replaced with a pointer:

```json
{
  "agent_id": "agent-abc-123",
  "type": "large_output",
  "timestamp": "2026-07-13T10:30:00+00:00",
  "truncated": true,
  "fetch_url": "/api/v1/cloud/agents/agent-abc-123/events/latest"
}
```

---

## Emitting Events (Server-Side)

Any backend service can emit events using the `AgentEventBus`:

```python
from app.services.agent_event_bus import AgentEventBus

bus = AgentEventBus(db=session)
bus.notify(
    agent_id="agent-abc-123",
    event_type="status_changed",
    payload={"status": "RUNNING"}
)
```

### Manual Emit (Testing)

```http
POST /api/v1/cloud/agents/agent-abc-123/events/emit
Authorization: Bearer <token>
Content-Type: application/json

{
  "event_type": "status_changed",
  "payload": {
    "status": "COMPLETED",
    "result": "Task finished successfully"
  }
}
```

---

## Common Event Types

| Type | Description |
|------|-------------|
| `connected` | Client connected to SSE stream |
| `keepalive` | Periodic ping (every 30s) |
| `status_changed` | Agent status transition (IDLE, RUNNING, PAUSED, ERROR) |
| `task_started` | Agent began executing a task |
| `task_completed` | Agent finished a task |
| `task_failed` | Agent task failed with error |
| `heartbeat` | Agent heartbeat signal |
| `log` | Agent log output |

---

## Channel Naming

Postgres NOTIFY channels follow the pattern `agent_events_{sanitized_id}` where dashes in the agent ID are replaced with underscores:

| Agent ID | Channel |
|----------|---------|
| `agent-abc-123` | `agent_events_agent_abc_123` |
| `my-agent` | `agent_events_my_agent` |
| `bot42` | `agent_events_bot42` |

The global admin channel is simply `agent_events`.

---

## Key Source Files

| File | Purpose |
|------|---------|
| `src/backend/app/services/agent_event_bus.py` | Core event bus: notify() + listen() + listen_all() |
| `src/backend/app/api/v1/endpoints/agent_events_sse.py` | SSE endpoints for per-agent and admin streams |
| `src/backend/tests/test_agent_event_bus.py` | Unit tests for notify and payload handling |