Event streams

A task streams while it runs: its reply text as it is written, the steps it takes, the media it makes, and its end. You receive it in one of two ways:

• Follow the session — GET /v1/sessions/{session_id}/events streams every task of the session, for as long as you stay connected.
• Stream on the request — send a message with "stream": true and the response is that one task's frames, ending with its end frame or after 10 minutes, whichever comes first.

Both send the same frames, as server-sent events.

Frames

Each frame has an id, an event name and JSON data, and is followed by a blank line:

id: 1760860912345-0:0
event: task.started
data: {"task_id": "exec-5a2c…"}

id: 1760860912400-0:0
event: text.delta
data: {"task_id": "exec-5a2c…", "delta": "Here is a vegetarian version"}

Every frame carries the task_id of the task it belongs to.

event	data	When
task.started	task_id	The task began.
text.delta	task_id, delta	More reply text. The deltas of a task add up to its final text.
task.progress	task_id, category, text	A step under way. category is search, image, video, music, memory, analysis or other; text is a short description for the user.
media.updated	task_id, slot, kind, status, attachment, error	An image, video, audio or file the task makes changed status (below).
ui.updated	task_id, ui	a2ui sessions only: the task's whole UI data, each time it changes. See A2UI replies.
suggestions	task_id, items	Up to three messages the user might send next. Comes just before the end frame.
task.completed	task_id, session_id, status, text, ui	The task finished. text is the whole reply; in an a2ui session ui is its UI data.
task.stopped	task_id, session_id, status, text, ui	The task was stopped. text is the reply so far.
task.failed	task_id, session_id, status, error	The task failed. error has a code and a message.

Every task ends with exactly one of task.completed, task.stopped or task.failed — this set never grows. Other event types may be added: ignore an event you do not recognize.

A frame can be large: an end frame carries the whole reply, and in an a2ui session its UI data.

Media

media.updated follows each image, video, audio clip or file a task makes, by its slot:

• status is running while it is being made, then one final state: completed, failed or canceled. Each state is sent once.
• kind is image, video, audio or file (null while it is not known yet).
• A completed one carries an attachment — type, filename, url, content_type and expires_at. The url is a download link that expires; the session's message list always has a fresh one.
• A failed one carries an error code, such as insufficient_credits or generation_failed.
• Some media take longer than their task: a media.updated frame may come after the task's end frame. Keep following the session to receive it.

Task failures

task.failed names why in error.code, from a short list that may grow: insufficient_credits (the project's balance ran out), timeout, input (the request or its media could not be used), upstream or network (a service the task depends on failed: retry later), and provision, config, system or unknown (something failed on our side). error.message is a sentence for the user.

Resuming: frame ids and Last-Event-ID

Every frame has its own id — an opaque string: store it and send it back, never parse it. Send one back in the Last-Event-ID header when you open the stream, and the stream continues right after that frame — nothing lost, nothing twice:

Last-Event-ID	The stream sends
A frame's id	Everything after that frame. Use it to reconnect after a dropped connection.
A task's last_event_id	That task from its first frame. Use it to follow a task you just started.
0	The whole session, from its start.
Absent, or an id the session no longer holds	Only what happens from now on. Earlier replies are in the message list.

Follow a task you started

POST /v1/sessions (with a first message) and POST /v1/sessions/{session_id}/messages return last_event_id: where the new task begins on the stream. Open the stream with it, and you receive the task from its first frame — even if the task started writing before you connected.

curl -N "$NEWEYES_API_BASE/v1/sessions/$SESSION_ID/events" \
  -H "Authorization: Bearer $NEWEYES_USER_TOKEN" \
  -H "Last-Event-ID: $LAST_EVENT_ID"

Reconnect

A connection lasts at most 10 minutes, and networks drop. Remember the id of the last frame you handled; when the connection ends, open the stream again with that id as Last-Event-ID.

Heartbeats

While there is nothing to send, the stream writes a comment line every 15 seconds:

: ping

If 45 seconds pass with nothing at all — no frame, no ping — treat the connection as dropped and reconnect with the last frame's id.

Stream on the request

Add "stream": true to a message, and its response is 200 with that task's own frames, from its first frame through its end frame, then the response ends:

curl -N "$NEWEYES_API_BASE/v1/sessions/$SESSION_ID/messages" -X POST \
  -H "Authorization: Bearer $NEWEYES_USER_TOKEN" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(uuidgen)" \
  -d '{"text": "Make it vegetarian.", "stream": true}'

• A refused message still answers with its JSON error and an error status, before any frame.
• The response holds only this task's frames. A media.updated frame that comes after the end frame is not on it: follow the session's stream for those.
• Like any event stream, the response lasts at most 10 minutes, and a task can take longer — a video, for example.
• A response that ends without the end frame does not stop the task — whether the connection dropped or the 10 minutes ran out. Continue on the session's stream with Last-Event-ID set to the last frame's id you received, or retry the request with the same Idempotency-Key to receive the same task again from its first frame.
• The response counts as one of the user's event streams.

Reading the stream in code

EventSource in browsers cannot send an Authorization header, so read the stream with fetch (or any HTTP client that streams the response body).

JavaScript

const END = ['task.completed', 'task.failed', 'task.stopped'];
const SILENCE_MS = 45_000; // three missed pings

// Follows one task to its end, reconnecting with the last frame's id when the connection drops
// or goes silent.
async function followTask(api, token, sessionId, taskId, startAt, onFrame) {
  let lastId = startAt;
  for (;;) {
    const silence = new AbortController();
    let timer = setTimeout(() => silence.abort(), SILENCE_MS);
    try {
      const response = await fetch(`${api}/v1/sessions/${sessionId}/events`, {
        headers: { Authorization: `Bearer ${token}`, 'Last-Event-ID': lastId },
        signal: silence.signal,
      });
      if (!response.ok) throw new Error((await response.json()).error.code);
      const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();
      let buffer = '';
      for (;;) {
        const { value, done } = await reader.read();
        if (done) break; // the connection ended: reconnect from lastId
        clearTimeout(timer);
        timer = setTimeout(() => silence.abort(), SILENCE_MS);
        buffer += value;
        let end;
        while ((end = buffer.indexOf('\n\n')) !== -1) {
          const frame = parse(buffer.slice(0, end));
          buffer = buffer.slice(end + 2);
          if (!frame) continue; // a ping
          lastId = frame.id;
          if (frame.data.task_id !== taskId) continue;
          onFrame(frame);
          if (END.includes(frame.event)) {
            await reader.cancel();
            return frame;
          }
        }
      }
    } catch (error) {
      if (error.name !== 'AbortError' && error.name !== 'TypeError') throw error;
      // silent too long (AbortError) or the network dropped (TypeError): reconnect from lastId
    } finally {
      clearTimeout(timer);
    }
  }
}

function parse(block) {
  const fields = {};
  for (const line of block.split('\n')) {
    if (!line || line.startsWith(':')) continue;
    const at = line.indexOf(':');
    fields[line.slice(0, at)] = line.slice(at + 1).replace(/^ /, '');
  }
  return fields.event ? { id: fields.id, event: fields.event, data: JSON.parse(fields.data) } : null;
}

Python

import json
import requests

END = {"task.completed", "task.failed", "task.stopped"}


def follow_task(api, token, session_id, task_id, start_at, on_frame):
    """Follows one task to its end, reconnecting from the last frame's id when dropped."""
    last_id = start_at
    while True:
        with requests.get(
            f"{api}/v1/sessions/{session_id}/events",
            headers={"Authorization": f"Bearer {token}", "Last-Event-ID": last_id},
            stream=True,
            timeout=(10, 45),  # 45 s without a byte (not even a ping): reconnect
        ) as response:
            response.raise_for_status()
            frame = {}
            try:
                for line in response.iter_lines(decode_unicode=True):
                    if line == "":
                        if "event" in frame:
                            last_id = frame["id"]
                            data = json.loads(frame["data"])
                            if data.get("task_id") == task_id:
                                on_frame(frame["event"], data)
                                if frame["event"] in END:
                                    return frame["event"], data
                        frame = {}
                    elif not line.startswith(":"):
                        name, _, value = line.partition(":")
                        frame[name] = value.removeprefix(" ")
            except (
                requests.exceptions.ConnectionError,  # silent too long, or connection lost
                requests.exceptions.ChunkedEncodingError,  # cut off mid-frame
            ):
                continue  # reconnect from last_id

Limits

An end user keeps at most 16 streams open at once, and at most 4 on one session. Opening another ends the oldest instead — that session's oldest, else the end user's oldest — and the stream it ends closes. A new stream is refused with 429 rate_limited only while every stream the end user holds was opened in the last 10 seconds.