Streaming | m8tes

Runs stream Server-Sent Events (SSE) by default. The SDK wraps them in a RunStream iterator.

Start a run, read text and tool events, then handle completion, failure, or cancellation.

Quick Patterns

Text only, live:

from m8tes import M8tes

client = M8tes()

for chunk in client.runs.stream_text(agent_id=1, message="Summarize open tickets", user_id="customer_42"):
    print(chunk, end="", flush=True)

import { M8tes } from "@m8tes/sdk";

const client = new M8tes(); // reads M8TES_API_KEY

const stream = client.runs.create({ agent_id: 1, message: "Summarize open tickets", user_id: "customer_42" });

for await (const chunk of stream.iterText()) process.stdout.write(chunk);

Text + post-run data (run_id, full text). Use with so the connection closes cleanly:

with client.runs.create(agent_id=1, message="Draft weekly report", user_id="customer_42") as stream:
    for chunk in stream.iter_text():
        print(chunk, end="", flush=True)
print(stream.text)    # all text deltas joined
print(stream.run_id)  # available after iteration

import type { RunStream } from "@m8tes/sdk";

const stream: RunStream = client.runs.create({ agent_id: 1, message: "Draft weekly report", user_id: "customer_42" });

for await (const chunk of stream.iterText()) process.stdout.write(chunk);

console.log(stream.output); // all text deltas joined
console.log(stream.runId); // the run id, set from the first event
console.log(stream.errors); // error messages the run emitted, if any

Just the final output (no streaming):

Python

run = client.runs.create_and_wait(agent_id=1, message="Generate report", user_id="customer_42")
print(run.output)

Follow-up replies (runs.reply()) stream the same way and inherit the run's settings. See Follow up on a run.

Event Types

These are the event.type values the Python SDK yields (it normalizes the raw Claude-native SSE frames into these friendlier names; see SSE Frame Format for the raw wire if you consume the HTTP stream directly).

Event	Description	Key fields
`text-delta`	Incremental text chunk	`event.delta`
`tool-call-start`	Agent starts using a tool	`event.tool_name`
`tool-call-delta`	Streaming tool arguments	`event.delta`
`tool-result-end`	Tool finished with result	`event.result`
`metadata`	Run metadata (includes run_id)	`event.payload["run_id"]`
`error`	Error during execution	`event.error`
`done`	Stream complete	`event.stop_reason` (`end_turn`, `max_tokens`, `stop_sequence`)
`run_metrics`	Execution stats	`event.execution_time_ms`, `event.input_tokens_used`

Filter events by class

from m8tes import TextDeltaEvent, ToolCallStartEvent

for event in client.runs.create(agent_id=1, message="Close tickets", user_id="customer_42"):
    if isinstance(event, TextDeltaEvent):
        print(event.delta, end="", flush=True)
    elif isinstance(event, ToolCallStartEvent):
        print(f"\nUsing tool: {event.tool_name}")

import type { M8tesStreamEvent } from "@m8tes/sdk";

// The events are a discriminated union: each case narrows to its own fields.

function render(event: M8tesStreamEvent) {
  switch (event.type) {
    case "text-delta":
      process.stdout.write(event.delta);
      break;
    case "tool-input-start":
      console.log("using tool:", event.toolName);
      break;
    case "tool-output-available":
      console.log("tool finished:", event.output);
      break;
    case "run-error":
      console.error(event.message);
      break;
  }
}

for await (const event of client.runs.create({ agent_id: 1, message: "close tickets", user_id: "customer_42" })) {
  render(event);
}

Detecting Failures

A run can fail mid-stream (expired credential, model rate limit, quota). The default iter_text() / stream.text path drops error events, so a failed run can otherwise look like an empty success.

The Python SDK also emits an error event for a terminal frame with subtype="success" and is_error=true; the subtype alone does not mean the run succeeded. This includes provider tool-count rejections. Read the persisted run's error_code and auto_retryable before retrying; tool_limit_exceeded is not automatically retryable. See Run error codes.

Opt into raising, or check after iterating:

# Raise RunFailedError if the run fails mid-stream
for event in client.runs.create(agent_id=1, message="...", raise_on_error=True, user_id="customer_42"):
    ...

# Or check without raising
with client.runs.create(agent_id=1, message="...", user_id="customer_42") as stream:
    for chunk in stream.iter_text():
        print(chunk, end="")
    if stream.has_errors:
        print("run failed:", stream.errors)

import { M8tes, RunFailedError } from "@m8tes/sdk";

const client = new M8tes();
// raiseOnError is the second argument: a run that emitted error events
// throws RunFailedError once iteration finishes.
const stream = client.runs.create({ agent_id: 1, message: "...", user_id: "customer_42" }, { raiseOnError: true });

try {
  for await (const chunk of stream.iterText()) process.stdout.write(chunk);
} catch (err) {
  if (!(err instanceof RunFailedError)) throw err;
  console.error("run failed:", stream.errors); // same messages, also readable without throwing
}

Reconnecting to a dropped stream

A stream can stop before the run does (proxy idle-timeout, network blip). The run keeps working on the server, so the SDK raises StreamInterruptedError with the run_id instead of ending as if the run were done. Wait for the result, and never send the request again: that starts a second run. To keep streaming instead, runs.stream(run_id) replays the run's full history then live deltas, so reset any local accumulation on reconnect; it returns 409 once the run is no longer executing (use runs.get(run_id) for the result).

Python

from m8tes import StreamInterruptedError

stream = client.runs.create(agent_id=1, message="long task", user_id="customer_42")
try:
    for chunk in stream.iter_text():
        print(chunk, end="", flush=True)
except StreamInterruptedError as exc:
    if exc.run_id is None:
        raise
    # A long task needs a wait budget past the 300 second default.
    run = client.runs.wait(exc.run_id, user_id="customer_42", timeout=1800)
    print(run.output)

The TypeScript SDK does the same. A stream that stops before the run does throws StreamInterruptedError with the runId, instead of ending as if the run were done. The run keeps working, so wait for its result and never send the request again:

TypeScript

import { M8tes, StreamInterruptedError } from "@m8tes/sdk";

const client = new M8tes();
const stream = client.runs.create({ agent_id: 1, message: "long task", user_id: "customer_42" });

try {
  for await (const chunk of stream.iterText()) process.stdout.write(chunk);
} catch (err) {
  if (!(err instanceof StreamInterruptedError) || err.runId === null) throw err;
  // A long task needs a wait budget past the 300 second default.
  const run = await client.runs.wait(err.runId, { user_id: "customer_42", timeout: 1800 });
  console.log(run.output);
}

To keep streaming instead, call client.runs.stream(err.runId): it replays the run from its start, and throws RunNotStreamingError once the run has finished (read the result with client.runs.get).

The server sends a keepalive every 15 seconds, so a long, quiet tool call does not trip the timeout. On a stream, timeout is the longest silence tolerated, never a limit on how long the run may take.

SSE Frame Format

Client	Events you receive
Python SDK or `@m8tes/react`	Normalized event types
Raw HTTP/SSE	Claude-native frames: `content_block_delta` for text; `content_block_start`, `tool_use`, and `tool_result` for tools

Raw wire example

data: {"type": "metadata", "run_id": 123, "mode": "chat", "sandbox_enabled": true}
data: {"type": "content_block_start", "id": "block_1", "block_type": "text"}
data: {"type": "content_block_delta", "id": "block_1", "delta": {"type": "text_delta", "text": "Here are "}}
data: {"type": "content_block_delta", "id": "block_1", "delta": {"type": "text_delta", "text": "the open tickets..."}}
data: {"type": "content_block_stop", "id": "block_1"}
data: {"type": "tool_use", "id": "tc_1", "name": "gmail_search", "input": {"query": "is:open"}}
data: {"type": "tool_result", "tool_use_id": "tc_1", "content": "...", "result": "..."}
data: {"type": "run_metrics", "execution_time_ms": 4200, "input_tokens_used": 1200, "completion_state": "complete"}
data: {"type": "done", "completion_state": "complete", "stop_reason": "end_turn", "message_count": 4}

Delegation (subagents) in the stream

An agent may delegate part of a job to a subagent: a worker that runs in its own context and reports a result back. You will see this as a tool call named Agent:

data: {"type": "tool_use", "id": "tc_2", "name": "Agent", "input": {"description": "Pull last week's spend", "subagent_type": "general-purpose", "prompt": "..."}}
data: {"type": "tool_result", "tool_use_id": "tc_2", "content": "...the subagent's report..."}

Treat it like any other tool call. The subagent's own steps are not streamed; you get the delegation and its result. Nothing about your event handling has to change.
Its tokens are part of the run. Delegated work is included in the run's cost and counts against the run's spending limits. run_metrics carries subagent_count and subagent_tokens so you can see how much of a run was delegated.
Permissions still apply. A subagent's tool calls go through the same approval rules as the agent's own. If a tool needs a human, it still asks, and the approval says which subagent asked for it.

The Python SDK normalizes these into the events (text-delta, tool-call-start, tool-result-end) shown in the table above.

Next: Runs · Human-in-the-Loop · Webhook Events