> ## Documentation Index
> Fetch the complete documentation index at: https://docs.melso.ai/llms.txt
> Use this file to discover all available pages before exploring further.

> ## Agent Instructions
> Melso product MCP: https://melso.ai/api/mcp (OAuth; one company per grant). REST: https://melso.ai. Documentation-search MCP: https://docs.melso.ai/mcp. The docs search server does not act on customer accounts.
>
> Use the live tool schema and respect company membership. Ask the person to authorize the exact fiscal document version in the signed-in app. A professional prepares and files included returns after authorization; AI does not file alone. A prepared file or approval receipt is not evidence of filing. Never claim completion from an accepted Task or Run. Keep credentials out of chat and logs. Agents features are internal or by private invitation.

# Streaming Task events

> Push a Task's changes to your integration over Server-Sent Events instead of polling.

<Info>Melso Agents is available only internally or by private invitation. Agents-specific features require existing access. For the autónomo fiscal service, see [Get started](/essentials/quickstart).</Info>

`GET /api/tasks/{id}/stream` pushes one Task's changes as [Server-Sent Events](https://html.spec.whatwg.org/multipage/server-sent-events.html): 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. Company API keys, personal access tokens and browser sessions can all open it.

Examples use the environment variables from [Authentication](/api-reference/authentication). UUID placeholders must be replaced with your own IDs.

## `GET /api/tasks/{id}/stream`

**Auth:** Exactly as `GET /api/tasks/{id}`: company membership and Area access. A company API key streams Tasks of its own company only. A Run token of an [isolated Task](/concepts/tasks-api#isolated-tasks) 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 company 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](/api-reference/errors), with the same statuses `GET /api/tasks/{id}` returns.

Set `TASK_ID` to your Task UUID or identifier and use the personal-token variables from [Authentication](/api-reference/authentication).

```bash theme={null}
export TASK_ID='YOUR_TASK_UUID'
curl --fail-with-body -sSN "$MELSO_URL/api/tasks/$TASK_ID/stream" \
  -H "Authorization: Bearer $MELSO_TOKEN" -H "X-Workspace-ID: $WORKSPACE_ID" -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`.

| Event | `id` | `data` | Sent when |
| - | - | - | - |
| `task` | cursor | TaskResponse, the `GET /api/tasks/{id}` body with `attachments` | The Task's response changes: status, lifecycle timestamps, title, run signals, attention, `unread`, metadata. |
| `run` | cursor | RunResponse, an item of `GET /api/tasks/{id}/runs` (including `status`, `error`, `failure_reason`, `attempt`, `max_attempts`, `parent_run_id`, `usage`) | A Run appears (a follow-up, retry, rollover, pause successor) and whenever its response changes: the [Run status](/concepts/overview#run-status) or usage changes. |
| `run.message` | cursor | RunMessagePayload, an item of `GET /api/runs/{runId}/messages` (`run_id`, `task_id`, `seq`, `type`, `tool`, `content`, `input`, `output`, `created_at`) | A Run's worker reports a trace message. |
| `task.message` | cursor | [TaskMessageResponse](/api-reference/tags/messages), an item of `GET /api/tasks/{id}/messages/page`, including `id`, `task_id`, `role`, `content` and optional `run_id` | A message enters the conversation: input, the Run's reply, a system message. |
| `interaction` | cursor | InteractionRequestResponse, the `GET /api/tasks/{id}/interactions/{interactionId}` body | A question or approval is created, answered, declined, cancelled or expires. |
| `stream.ready` | none | `{"task_id":string,"resumed":boolean}` | The snapshot or replay has been sent; live changes follow. |
| `stream.end` | none | `{"reason":string}` | The server is about to close the stream; see [Closing and reconnecting](#closing-and-reconnecting). |

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

```text theme={null}
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:

1. New `run` events, each followed by its `run.message` trace.
2. `run.message` events for Runs you already have.
3. Conversation messages (`task.message`).
4. Questions and approvals (`interaction`).
5. State changes of Runs you already have (`run`).
6. The updated `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: do not parse or build it. It does not expire and belongs to its Task. It is not a credential: resuming requires both the cursor and a credential with current access to that Task. Persist it with the Task ID in your application's private state to resume after a restart; keep cursors and Task IDs out of analytics logs.

## Closing and reconnecting

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

| Reason | Meaning | What to do |
| - | - | - |
| `max_duration` | The stream reached 25 minutes. | Reconnect with `Last-Event-ID`. |
| `shutdown` | The server instance is restarting. | Reconnect with `Last-Event-ID`. |
| `lagging` | The stream fell too far behind the live changes. | Reconnect with `Last-Event-ID`; the replay catches up. |
| `unavailable` | A temporary server failure. | Reconnect with `Last-Event-ID` after a short backoff. |
| `task_unavailable` | The Task was deleted, or the credential lost access to it. | Stop; a reconnect answers 404. |
| `credential_invalid` | The credential was revoked or expired (a Run token also expires when its Run ends). | Stop; use a valid credential. |

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 company API key's [rate limit](/api-reference/rate-limits); 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](#websocket).

## Example (Node.js 22)

Follows one Task until it closes, resuming across disconnects. The [message contract](/concepts/messages-interactions) identifies replies with `role: "assistant"`; the example reads that field. Replace `handle` with your own logic and persist the cursor after handling an event. The example retains it in memory; saving it durably with your handled state lets you resume after a process restart.

```typescript theme={null}
const base = process.env.MELSO_URL ?? "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

Company 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 value of `MELSO_API_KEY` as the first frame (replace the placeholder below). Never put a key in the URL. Company identifiers are not credentials, but keep company-specific connection URLs out of shared links and analytics logs:

```json theme={null}
{"type":"auth","payload":{"token":"YOUR_MELSO_API_KEY"}}
```

The server answers `{"type":"auth_ack"}`. A company key receives allowlisted Task, Run, conversation-message and reaction notifications, Area changes, company updates/deletion and access-loss notifications, subject to its Task and Area access. Interaction changes arrive as Task updates; read the Task's interactions to inspect them. Other member events, skills and company-setting events are excluded. A key naming another company 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.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.