Streaming events - Clatter Labs Docs

GET /v1/runs/{id}/events returns a text/event-stream. Any SSE client works, or read the response body line by line.

Event types

Event	Data	When
`status`	`{ "status": "running" }`	The run's status changed
`step`	A step object	Cosmos wrote a message or used a tool
`done`	The final run	The run finished (completed, failed or cancelled)
`timeout`	`{ "message", "after" }`	The connection reached 10 minutes; the run continues
`error`	`{ "code", "message" }`	The stream had to stop, for example the key was revoked

A step is either a message or a tool call:

JSON

{ "index": 0, "type": "message", "text": "I'll scaffold the project first." }
{ "index": 1, "type": "tool", "name": "computer_terminal", "ok": true,
  "args": "{\"cmd\":\"npm init -y\"}", "output": "Wrote to package.json" }

text is capped at 4,000 characters, args at 400 and output at 600. The full run is always available from GET /v1/runs/{id}.

Reconnecting

One stream connection lasts up to 10 minutes. Count the step events you've received and reconnect with ?after= that number; the stream continues from the next step.

async function follow(runId) {
  let after = 0;
  for (;;) {
    const res = await fetch(`${BASE}/v1/runs/${runId}/events?after=${after}`, { headers });
    const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
    let buf = "", done = false;
    for (;;) {
      const { value, done: end } = await reader.read();
      if (end) break;
      buf += value;
      let i;
      while ((i = buf.indexOf("\n\n")) >= 0) {
        const chunk = buf.slice(0, i); buf = buf.slice(i + 2);
        const event = /^event: (.*)$/m.exec(chunk)?.[1];
        const data = JSON.parse(/^data: (.*)$/m.exec(chunk)?.[1] ?? "null");
        if (event === "step") { after = data.index + 1; console.log(data); }
        if (event === "done") { done = true; return data; }
      }
    }
    if (done) break; // otherwise: timeout or dropped connection, reconnect
  }
}
import json, requests

def follow(run_id):
    after = 0
    while True:
        with requests.get(f"{BASE}/v1/runs/{run_id}/events",
                          params={"after": after}, headers=headers, stream=True) as r:
            event = None
            for line in r.iter_lines(decode_unicode=True):
                if line.startswith("event: "):
                    event = line[7:]
                elif line.startswith("data: "):
                    data = json.loads(line[6:])
                    if event == "step":
                        after = data["index"] + 1
                        print(data)
                    elif event == "done":
                        return data
        # timeout or dropped connection: loop and reconnect with ?after=

The stream re-checks your key every few minutes. If the key is revoked while you're connected you get an error event and the stream ends.