Thread streams

A thread stream carries an agent's output across runs on the same thread. You subscribe once to receive model text, tool calls and results, approval requests, and errors as chunks. The subscription stays open between runs, so you don't need to create a new response stream each time you send input.

You can also request stored messages at the start of a subscription. This lets your application show the conversation so far before displaying new output.

Quickstart

Subscribe before you send input so you can receive output from the start of the run. This example uses subscribeToThread() from the JavaScript client SDK. It needs a running Mastra server with a support-agent agent. Configure Memory on that agent if you want to store history.

For server-side code that uses the Agent instance directly, follow the Agent quickstart instead.

src/app/chat.ts

import { MastraClient } from '@mastra/client-js'

const client = new MastraClient({ baseUrl: 'http://localhost:4111' })
const agent = client.getAgent('support-agent')

const subscription = await agent.subscribeToThread({
  resourceId: 'user-123',
  threadId: 'thread-456',
})

await agent.sendMessage({
  message: 'Show the shorter version.',
  resourceId: 'user-123',
  threadId: 'thread-456',
})

await subscription.processDataStream({
  onChunk: chunk => {
    console.log(chunk)
  },
  reconnect: true,
})

client.getAgent() selects the server's agent. sendMessage() returns when Mastra accepts the input. processDataStream() passes output to onChunk, which prints each chunk here. It continues listening after the answer finishes.

Read the stream

Stored history and live parts

Pass withInitialHistory: true to begin with a thread-history chunk, which carries the newest page of stored messages in oldest-first order. perPage defaults to 40. Subscribers then receive retained parts newer than that history, followed by live parts. History requires memory. Without it, or before the thread has messages, the chunk still arrives first with an empty messages array.

Chunks and runs

Use runId to group chunks from the same run. The exception is thread-history, which has an empty runId because its messages can span several runs. The subscription's activeRunId() returns the active run's ID or null.

The public stream uses AgentChunkType, with ThreadHistoryChunk for stored history. The ChunkType reference lists the fields for model output, tool calls and results, approvals, aborts, and errors. To observe queue counts, use subscribeThreadEvents() separately. Internal runtime events aren't part of these public APIs.

Approve tool calls

When a run pauses for tool approval, send your decision with client.getAgent().sendToolApproval() or the server approval route. The existing subscription receives output when the run resumes.

A paused run keeps ownership of the thread. Signals that try to start a new run receive an accepted action of blocked until the paused run resumes or is aborted. Pending approvals and suspensions are replayed when you subscribe again. Completed runs aren't. When the agent sees your input covers the interrupt and wait rules for input that arrives during a run.

Filter signals in a subscription

hideSignals filters signal chunks for one subscriber. Pass true to hide every recognized signal, false to show them, or an array of signal types. It doesn't change model context, stored messages, shared broadcasts, or other subscribers, and it isn't a security boundary. The option exists on the in-process core subscription. HTTP and client-js subscription requests don't support it.

Stop a run or close a subscription

abort() requests cancellation of the active run. unsubscribe() closes the subscription without aborting, so the run continues and you can subscribe again later. Pending signals survive an abort unless you pass clearPendingSignals: true.

Reconnect

Use reconnect: true for long-lived subscriptions. processDataStream() then resubscribes when the transport closes or a reconnect request fails, retrying indefinitely with a one-second delay by default. A 4xx other than 408 or 429, or a 501, stops reconnecting and rejects. See the subscribeToThread() reference for retry options.

A reconnect can resume output that the shared PubSub backend still retains. It can't recover events held only in another process's in-memory cache. Use withInitialHistory to reload the newest page of stored messages when you need conversation history.

Stored messages aren't a recording of every streamed text fragment. Reconnecting doesn't replay completed stored runs or reconstruct fragments that weren't retained.

SSE heartbeats for custom endpoints

If you expose your own Server-Sent Events (SSE) endpoint for thread subscriptions, send periodic heartbeat frames while the stream is idle. This keeps browsers, proxies, and load balancers from closing the connection before the next signal or model chunk arrives.

The following example sends an SSE comment every 25 seconds:

src/api/subscribe.ts

const heartbeat = setInterval(() => {
  controller.enqueue(encoder.encode(': keep-alive\n\n'))
}, 25_000)

request.signal.addEventListener('abort', () => {
  clearInterval(heartbeat)
})

Use heartbeats together with client-side reconnect logic. Heartbeats reduce idle disconnects, while reconnects recover when the network or runtime still closes the stream.