Streaming Task events

GET /api/tasks/{id}/stream pushes one Task's changes as Server-Sent Events: the Task, its Runs, their trace messages, the conversation and interactions. It replaces polling GET /api/tasks/{id}, /runs, /messages/page and /interactions. Every event carries a cursor as its id, so a client that reconnects with Last-Event-ID receives exactly what it missed, once. Workspace API keys, personal access tokens and browser sessions can all open it.

Examples use the environment variables from Overview. UUID placeholders must be replaced with your own IDs.

GET /api/tasks/{id}/stream

Auth: Exactly as GET /api/tasks/{id}: workspace membership and Area access. A workspace API key streams Tasks of its own workspace only. A Run token of an isolated Task streams only that Task, and a Run token of another Task never sees an isolated one.

Request: Path: Task UUID or identifier. Headers: Last-Event-ID (optional) resumes after that event. Query: cursor resumes like Last-Event-ID for clients that cannot set the header (the header wins); from=start replays the Task's history on a stream opened without a cursor (default from=now). A personal access token or browser session selects the workspace as on every route; a browser EventSource, which cannot set headers, uses ?workspace_id= or ?workspace_slug= with its session cookie.

Response: 200 with Content-Type: text/event-stream, Cache-Control: no-cache, no-transform and X-Accel-Buffering: no, then events until the server closes the stream.

Status: 200; 400 invalid cursor, a cursor issued for another Task, or an unknown from; 429 too many open streams or requests (honor Retry-After); 503 live updates unavailable. Authentication and resource-access errors follow Overview, with the same statuses GET /api/tasks/{id} returns.

curl -sSN "$MELSO_URL/api/tasks/$ID/stream" \
  -H "Authorization: Bearer $MELSO_TOKEN" -H "Accept: text/event-stream"

Events

Each event's data is one line of JSON with the same shape the REST read returns, so the same parsing code serves both. Apply events as upserts: replace the Task, and key Runs, conversation messages and interactions by id, trace messages by run_id and seq.

The stream also sends retry: 2000 (a reconnect delay hint for EventSource) and a : ping comment every 15 seconds.

event: run
id: eyJ2IjoxLCJrIjoi…
data: {"id":"4f5c…","task_id":"9b1e…","status":"completed","attempt":1,"failure_reason":"",…}

event: stream.end
data: {"reason":"max_duration"}

Order

A write usually changes several objects. The stream sends them in this order: new Runs, each followed by its trace; trace messages of Runs you already have; conversation messages; interactions; state changes of Runs you already have; then the Task. A finished Run therefore arrives after its reply and trace, and the Task's new status after both. A Run's trace messages arrive in seq order, except that a message its worker persisted late arrives after higher ones: order a trace by seq.

What is not streamed

Edits, deletions and reactions of conversation messages; Run progress summaries; Run activity events (GET /api/runs/{runId}/events). Read those from their REST routes when you need them.

A follow-up sent while another Run is working stays out of the conversation until its turn, as in GET /api/tasks/{id}/messages/page. The stream sends it when the conversation shows it, usually placed right after the reply it waited for, so its created_at can be later than the moment it was sent. A message that moves within the conversation this way is sent again with its new created_at.

Opening, resuming and history

Opening without a cursor sends a snapshot: task, every Run (oldest first), every pending interaction, then stream.ready, then live changes. History (earlier trace messages, conversation messages, resolved interactions) is not replayed; read it once from the REST routes.

from=start also replays the history: each Run followed by its whole trace, every visible conversation message oldest first, and every interaction, then stream.ready.

Resuming. The id of every Task, Run, message and interaction event is a cursor that describes everything the stream had sent up to and including that event. After processing an event, keep its id. To resume, send the last one as Last-Event-ID (a browser EventSource does this automatically when it reconnects) or as ?cursor=. The stream first sends everything that changed since that event, including changes made while you were away, then stream.ready, then live changes. Nothing is sent twice and nothing is skipped, including rows that were written out of order. A resume sends each changed object's current state, not every intermediate one: a Run that went from running to completed while you were away arrives once, as completed.

The cursor is opaque base64url: do not parse or build it. It does not expire and belongs to its Task. It is usually a few hundred bytes and grows with the Runs still in progress and the messages of the last ten minutes. Persist it next to the Task ID to resume after a restart.

Closing and reconnecting

A stream never runs longer than 25 minutes. Before closing, the server sends stream.end with a reason:

A connection that ends without stream.end, or that stays silent for 45 seconds (three missed pings), dropped: reconnect with Last-Event-ID, backing off exponentially after repeated failures. Access and the credential are re-checked every minute and whenever Area access changes, so revoking a key or removing someone from an Area closes the streams it opened within about a minute.

Opening a stream counts as one request against a workspace API key's rate limit; events do not count. One identity can hold up to 50 open streams per API instance; beyond that the stream answers 429 with Retry-After. To follow many Tasks at once, open one stream per Task or use the WebSocket.

Example (Node.js 22)

Follows one Task until it closes, resuming across disconnects. Replace handle with your own logic; the cursor is saved only after an event was handled, so a crash replays it instead of losing it.

const base = process.env.MELSO_API_BASE ?? "https://melso.ai";
const key = process.env.MELSO_API_KEY!; // workspace API key (mwk_…)
const taskId = process.env.MELSO_TASK_ID!;
let cursor = process.env.MELSO_STREAM_CURSOR ?? ""; // persist it with the Task ID

function handle(event: string, data: any) {
  if (event === "run") console.log("Run", data.id, data.status, data.failure_reason ?? "");
  if (event === "task.message" && data.role === "assistant") console.log("Reply:", data.content);
  if (event === "interaction" && data.status === "pending") console.log("Needs input:", data.message);
  if (event === "task" && (data.completed_at || data.cancelled_at)) return "closed";
}

for (let failures = 0; ; ) {
  const abort = new AbortController();
  let silence = setTimeout(() => abort.abort(), 45_000);
  let stop = false;
  try {
    const headers: Record<string, string> = { Authorization: `Bearer ${key}`, Accept: "text/event-stream" };
    if (cursor) headers["Last-Event-ID"] = cursor;
    const res = await fetch(`${base}/api/tasks/${taskId}/stream`, { headers, signal: abort.signal });
    if ([400, 401, 403, 404].includes(res.status)) throw Object.assign(new Error(await res.text()), { fatal: true });
    if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);
    let buffer = "";
    const decoder = new TextDecoder();
    for await (const chunk of res.body) {
      clearTimeout(silence);
      silence = setTimeout(() => abort.abort(), 45_000);
      buffer += decoder.decode(chunk, { stream: true });
      for (let end = buffer.indexOf("\n\n"); end >= 0; end = buffer.indexOf("\n\n")) {
        const block = buffer.slice(0, end);
        buffer = buffer.slice(end + 2);
        let event = "", id = "", data = "";
        for (const line of block.split("\n")) {
          if (line.startsWith("event: ")) event = line.slice(7);
          else if (line.startsWith("id: ")) id = line.slice(4);
          else if (line.startsWith("data: ")) data += line.slice(6);
        }
        if (!event) continue; // a ping or the retry hint
        const payload = JSON.parse(data);
        if (event === "stream.end") {
          stop = payload.reason === "task_unavailable" || payload.reason === "credential_invalid";
          continue;
        }
        if (handle(event, payload) === "closed") stop = true;
        if (id) cursor = id;
      }
      if (stop) break;
    }
    failures = 0;
  } catch (error: any) {
    if (error?.fatal) throw error;
    failures++;
  } finally {
    clearTimeout(silence);
    abort.abort();
  }
  if (stop) break;
  await new Promise((resolve) => setTimeout(resolve, Math.min(30_000, 1_000 * 2 ** failures)));
}

WebSocket

Workspace API keys can also use the realtime WebSocket the web app uses, which carries changes of every Task the key can read. Connect to wss://melso.ai/ws?workspace_id=<workspace-uuid> (or ?workspace_slug=<slug>), then send the key as the first frame:

{"type":"auth","payload":{"token":"mwk_…"}}

The server answers {"type":"auth_ack"} and the connection receives the Task, Run, message, interaction, Area and workspace frames a member of that workspace receives, filtered by the key's Task and Area access. Frames about members, skills and other workspace settings, which a key cannot read over REST either, are not sent. A key naming another workspace receives {"error":"this API key belongs to a different workspace","code":"workspace_key_scope"} and is disconnected; an invalid, revoked or expired key receives {"error":"invalid, expired or revoked API key"}. The key is re-checked every minute: revoking it closes the connection with status 1008.

WebSocket frames are change notifications ({"type":"task:updated","payload":{…},"task_id":…}, run:completed, run:message, message:created, …) without cursors or replay, and their payloads are the web app's internal shapes. Use them as triggers to read the REST routes, or use the SSE stream above when you need a lossless, resumable feed of one Task.