SSE Endpoints

Server-Sent Events (SSE) provide a lightweight, unidirectional streaming mechanism for receiving real-time call updates in the dashboard. Unlike WebSockets, SSE uses standard HTTP and automatically reconnects on connection loss.

SSE endpoints are designed for the Agent Studio dashboard UI. For backend integrations, use Webhooks instead.

Stream All Call Events

GET /api/v1/sse/calls/stream

Stream real-time events for all calls in your tenant. Useful for dashboards showing overall call activity.

Headers:

Header	Required	Description
`Authorization`	Yes	Bearer token or API key

Events:

Event	Description	Data
`connected`	Connection established	`{ message, tenant_id }`
`active_count`	Active call count changed	`{ count, timestamp }`
`calls_update`	Call list updated	`{ calls: [...], timestamp }`
`heartbeat`	Keepalive ping	`{ timestamp }`
`error`	Error occurred	`{ message, timestamp }`

Example Response Stream:

event: connected
data: {"message":"Connected to call events stream","tenant_id":"uuid"}

event: active_count
data: {"count":3,"timestamp":"2026-01-20T10:30:00Z"}

event: calls_update
data: {"calls":[{"id":"uuid","status":"active",...}],"timestamp":"2026-01-20T10:30:00Z"}

event: heartbeat
data: {"timestamp":"2026-01-20T10:30:02Z"}

Stream Single Call Events

GET /api/v1/sse/calls/{call_id}/stream

Stream real-time events for a specific call. The stream automatically closes when the call ends.

Path Parameters:

Parameter	Type	Description
`call_id`	uuid	Call ID to monitor

Headers:

Header	Required	Description
`Authorization`	Yes	Bearer token or API key

Events:

Event	Description	Data
`connected`	Connection established	`{ message, call_id }`
`status`	Call status changed	`{ call_id, status, status_reason, duration_seconds, timestamp }`
`transcript`	New transcript entry	`{ call_id, role, content, agent, timestamp }`
`agent_changed`	Agent handoff occurred	`{ call_id, agent, previous_agent, timestamp }`
`ended`	Call ended (stream closes)	`{ call_id, final_status }`
`heartbeat`	Keepalive ping	`{ timestamp }`
`error`	Error (e.g., call not found)	`{ message, call_id }`

Example Response Stream:

event: connected
data: {"message":"Connected to call uuid events","call_id":"uuid"}

event: status
data: {"call_id":"uuid","status":"active","timestamp":"2026-01-20T10:30:00Z"}

event: transcript
data: {"call_id":"uuid","role":"assistant","content":"Hello!","agent":"greeter","timestamp":"2026-01-20T10:30:05Z"}

event: transcript
data: {"call_id":"uuid","role":"user","content":"Hi there","agent":"greeter","timestamp":"2026-01-20T10:30:08Z"}

event: agent_changed
data: {"call_id":"uuid","agent":"meal-agent","previous_agent":"greeter","timestamp":"2026-01-20T10:30:15Z"}

event: ended
data: {"call_id":"uuid","final_status":"completed"}

Client Implementation

JavaScript (Browser)

const token = 'your-jwt-token';

// Note: EventSource doesn't support custom headers natively
// Use a polyfill like eventsource-polyfill or fetch-event-source

import { fetchEventSource } from '@microsoft/fetch-event-source';

await fetchEventSource('/api/v1/sse/calls/stream', {
  headers: {
    'Authorization': `Bearer ${token}`,
  },
  onopen(response) {
    console.log('Connected');
  },
  onmessage(event) {
    const data = JSON.parse(event.data);
    
    switch (event.event) {
      case 'active_count':
        updateActiveCallsDisplay(data.count);
        break;
      case 'calls_update':
        updateCallsList(data.calls);
        break;
    }
  },
  onerror(err) {
    console.error('SSE error:', err);
  },
});

Python

import httpx
import json

def stream_calls(api_key: str):
    """Stream all call events."""
    url = "https://api.example.com/api/v1/sse/calls/stream"
    headers = {"Authorization": f"Bearer {api_key}"}
    
    with httpx.stream("GET", url, headers=headers, timeout=None) as response:
        for line in response.iter_lines():
            if line.startswith("event:"):
                event_type = line[7:].strip()
            elif line.startswith("data:"):
                data = json.loads(line[5:])
                handle_event(event_type, data)

def handle_event(event_type: str, data: dict):
    if event_type == "active_count":
        print(f"Active calls: {data['count']}")
    elif event_type == "calls_update":
        for call in data["calls"]:
            print(f"Call {call['id']}: {call['status']}")

cURL

curl -N -H "Authorization: Bearer YOUR_TOKEN" \
  https://api.example.com/api/v1/sse/calls/stream

Connection Behavior

Polling Interval: Events are checked every 2 seconds for all-calls stream, 1 second for single-call stream
Heartbeat: Sent with each poll cycle to keep connection alive
Auto-close: Single-call streams close automatically when call ends
Reconnection: Clients should implement reconnection logic on connection loss

SSE vs Webhooks

Use Case	Recommended
Dashboard real-time updates	SSE
Backend event processing	Webhooks
External system integration	Webhooks
Audit logging	Webhooks

SSE is used internally by the Agent Studio dashboard for real-time UI updates. For external backend integrations, always use Webhooks which provide reliable delivery with retry logic.

SSE Endpoints

On this page