Streaming & SSE

Thallus uses Server-Sent Events (SSE) to stream real-time progress updates during chat queries. Instead of waiting for a complete response, your client receives a stream of events as the system plans, runs agents, calls tools, and generates the final answer.

How streaming works

The streaming flow has three steps: start a task, subscribe to progress events, then fetch the final result.

1. Start a task

POST /api/v1/chat/stream
Authorization: Bearer <token>
Content-Type: application/json

{
  "message": "What were our top-selling products last quarter?",
  "conversation_id": "optional-uuid",
  "mode": "auto"
}

The server responds immediately with a task ID:

{
  "task_id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "processing"
}

Only one task can be active per user at a time. Starting a new task while one is running returns a 409 Conflict.

2. Subscribe to progress

Open an SSE connection to receive real-time updates:

GET /api/v1/chat/{task_id}/progress?last_id=0
Authorization: Bearer <token>

The last_id parameter controls where to start reading. Use 0 for the beginning, or the last received stream ID for reconnection.

3. Fetch the result

After receiving a complete event, fetch the final response:

GET /api/v1/chat/{task_id}/result
Authorization: Bearer <token>

{
  "task_id": "550e8400-...",
  "status": "completed",
  "response": "Based on your sales data...",
  "citations": [...],
  "conversation_id": "..."
}

SSE wire format

Each event follows the standard SSE format with an id field (the stream position) and a data field containing JSON:

The server sends periodic heartbeat events during idle periods to keep the connection alive. Heartbeat events should be ignored by your client.

Events are stored server-side with a retention window after task completion, enabling reconnection and replay.

Event types

Progress events are grouped into six categories. Every event includes the base fields: task_id, update_type, message, event_id, and timestamp. Additional fields vary by event type.

Planning events

Agent events

Tool events

Parallel execution events

Streaming events

General events

The complete event is the terminal event. When your client receives it, close the SSE connection and fetch the final result.

Event payload fields

Every event includes these base fields:

Context fields (present when applicable):

Client example

This example uses a fetch-based SSE library to connect with a Bearer token (standard EventSource does not support custom headers):

import { fetchEventSource } from "fetch-event-source-library";

const taskId = "550e8400-...";
let lastEventId = "0";
const seenEvents = new Set();

await fetchEventSource(`/api/v1/chat/${taskId}/progress?last_id=${lastEventId}`, {
  headers: {
    Authorization: `Bearer ${accessToken}`,
  },
  onmessage(event) {
    if (event.event === "heartbeat") return;

    const data = JSON.parse(event.data);

    // Deduplicate on reconnection
    if (seenEvents.has(data.event_id)) return;
    seenEvents.add(data.event_id);

    // Track position for reconnection
    lastEventId = event.id;

    if (data.update_type === "text_delta") {
      appendToResponse(data.message);
    } else if (data.update_type === "complete") {
      fetchResult(taskId);
    }
  },
  onerror(err) {
    // Return to trigger reconnection with backoff
    // fetch-event-source handles this automatically
  },
});

Reconnection

If the SSE connection drops, reconnect by passing the last received stream ID in the last_id query parameter. The server replays any events you missed.

Recommended strategy

• Maximum attempts: 10
• Backoff: Exponential — 1s, 2s, 4s, 8s, 16s, 30s (capped at 30s)
• Resume from: The id field of the last received SSE event

Connection states

Deduplication

On reconnection, you may receive events that your client already processed. Use the event_id field (a UUID unique to each event) to deduplicate. The recommended approach is an LRU cache of recent event IDs with a cap around 500 entries.

Recovery endpoints

If your client loses context entirely (e.g., page refresh), use these endpoints to recover:

Page refresh recovery flow

1. Call GET /chat/active to check for in-flight tasks
2. If active_task is present, reconnect SSE to that task ID
3. If completed_task is present, fetch the result directly
4. If neither, no recovery needed

Limits

Related pages

• Core API Endpoints — full endpoint reference including the chat streaming flow
• Error Handling — error responses during streaming (401, 404, 429)
• Rate Limiting — SSE connection limits and request throttling
• How Orchestration Works — understanding plan and agent execution stages