View as MarkdownPlain text for this page Open in ChatGPTQuestions about this page Open in ClaudeQuestions about this page

Runtime streams

Exchange live bytes with the execution host of a thread

What is a runtime stream?

A runtime stream is a live, two-way byte stream between agent code and the thread's execution host. The execution host is the client that runs work for the thread outside the Worker, such as a process on a workstation.

Agent code opens a stream with state.runtimeStreams.open(). The runtime carries the bytes and does not read them. The capability name and the metadata have meaning only to your agent code and your execution host.

Use a runtime stream when a tool must send or receive a large or continuous byte sequence, such as media, an archive, or the output of a long process.

Feature detection

runtimeStreams is an optional member of ThreadState in the Standard Agent Specification. Standard Agents Builder provides it on every ThreadState. Check for it before use, so the same code runs on a runtime that omits it:

const runtimeStreams = state.runtimeStreams;
if (!runtimeStreams) {
  return { status: 'error', error: 'This runtime has no runtime streams.' };
}

Opening a stream

const opened = await runtimeStreams.open({
  capability: 'workstation_transcoder',
  metadata: { inputFormat: 'wav', outputFormat: 'opus' },
  timeoutMs: 10_000,
});

Property	Type	Description
`capability`	`string`	The name of the host capability to open
`metadata`	`Record<string, unknown>`	Configuration that the runtime delivers to the host unchanged
`timeoutMs`	`number`	The time limit for opening the stream, measured from the call. It covers waiting for a disconnected owner to reconnect and waiting for the client to accept. Standard Agents Builder uses 30 seconds when you omit it

metadata and timeoutMs are optional.

open() does not throw for a refused or unavailable stream. It resolves to a result:

type RuntimeStreamResult =
  | { ok: true; stream: RuntimeByteStream }
  | { ok: false; error: string; code?: RuntimeStreamErrorCode };

Failure codes

Code	Meaning in Standard Agents Builder
`unavailable`	No execution host is connected to the thread, or the host could not be reached
`refused`	The execution host answered the request with an error
`timeout`	The execution host did not accept the stream within the time limit
`unsupported`	A parent thread runs a runtime version that cannot carry the stream

When the thread has a recorded execution host whose connection is closed, Standard Agents Builder signals that host to reconnect and waits up to the time limit. When the thread has no recorded execution host, open() returns unavailable at once.

The stream

interface RuntimeByteStream {
  readonly readable: ReadableStream<Uint8Array>;
  readonly writable: WritableStream<Uint8Array>;
  readonly closed: Promise<{ ok: boolean; error?: string }>;
  close(reason?: string): void;
}

readable carries bytes from the execution host to your code.
writable carries bytes from your code to the execution host.
Bytes arrive in the order they were written, in both directions.
Closing writable ends your side of the stream. The host can continue to send bytes through readable.
close(reason?) aborts the stream in both directions. A second call does nothing.
closed settles once with the final outcome of the stream.

Lifetime

A runtime stream is a live resource. Standard Agents Builder does not store it:

A stream ends with an error when the execution host disconnects.
A stream does not continue after the thread's Durable Object restarts.
A stream opened by a child thread travels through the nearest parent thread that has an execution host.

Call stream.close() in a finally block, so a stream never stays open after your code returns.

Example

This tool sends a file from the thread's file system to a host transcoder and stores the result:

// agents/tools/transcode_audio.ts
import { defineTool } from '@standardagents/spec';
import { z } from 'zod';

export default defineTool({
  description: 'Transcode a WAV file from the thread file system to Opus.',
  args: z.object({
    path: z.string().describe('Path of the WAV file'),
  }),
  execute: async (state, args) => {
    const runtimeStreams = state.runtimeStreams;
    if (!runtimeStreams) {
      return { status: 'error', error: 'This runtime has no runtime streams.' };
    }

    const source = await state.readFile(args.path);
    if (!source) {
      return { status: 'error', error: `File not found: ${args.path}` };
    }

    const opened = await runtimeStreams.open({
      capability: 'workstation_transcoder',
      metadata: { inputFormat: 'wav', outputFormat: 'opus' },
      timeoutMs: 10_000,
    });
    if (!opened.ok) {
      return { status: 'error', error: opened.error };
    }

    const { stream } = opened;
    try {
      const writer = stream.writable.getWriter();
      try {
        await writer.write(new Uint8Array(source));
        await writer.close();
      } finally {
        writer.releaseLock();
      }

      const chunks: Uint8Array[] = [];
      let size = 0;
      const reader = stream.readable.getReader();
      try {
        for (;;) {
          const { value, done } = await reader.read();
          if (done) break;
          chunks.push(value);
          size += value.byteLength;
        }
      } finally {
        reader.releaseLock();
      }

      const outcome = await stream.closed;
      if (!outcome.ok) {
        return { status: 'error', error: outcome.error ?? 'The transcoder stream failed.' };
      }

      const output = new Uint8Array(size);
      let offset = 0;
      for (const chunk of chunks) {
        output.set(chunk, offset);
        offset += chunk.byteLength;
      }
      const target = args.path.replace(/\.wav$/, '.opus');
      await state.writeFile(target, output.buffer, 'audio/ogg');
      return { status: 'success', result: `Wrote ${target} (${size} bytes).` };
    } finally {
      stream.close();
    }
  },
});

Next steps

Tools

Open runtime streams from tool code

ThreadState

See the other members of the state object