Streams

Stream methods live on world.streams (the streams sub-object of the World instance returned by await getWorld()). Use them to write chunks, read streams, and manage stream lifecycle outside of the standard getWritable() pattern.

For most streaming use cases, use getWritable() inside steps. Direct stream methods are for advanced scenarios like building custom stream consumers or managing streams from outside a workflow.

Import

import { getWorld } from "workflow/runtime";

const world = await getWorld();
// Stream methods are called on world.streams, e.g. world.streams.write()

Methods

write()

Write a data chunk to a named stream.

await world.streams.write(runId, "default", chunk);

Parameters:

Parameter	Type	Description
`runId`	`string`	The workflow run ID
`name`	`string`	The stream name
`chunk`	`string \| Uint8Array`	Data to write

writeMulti()

Write multiple chunks in a single operation. Optional optimization: not all World implementations support it. Falls back to sequential write() calls if unavailable.

await world.streams.writeMulti?.(runId, "default", [chunk1, chunk2]);

Parameters:

Parameter	Type	Description
`runId`	`string`	The workflow run ID
`name`	`string`	The stream name
`chunks`	`(string \| Uint8Array)[]`	Chunks to write, in order

get()

Read data from a named stream as a live ReadableStream that waits for new chunks in real time.

const readable = await world.streams.get(runId, "default");

Parameters:

Parameter	Type	Description
`runId`	`string`	The workflow run ID
`name`	`string`	The stream name
`startIndex`	`number`	Optional. Positive values skip chunks from the start (0-based). Negative values read from the tail (e.g. `-3` starts 3 chunks from the end). Clamped to 0.

Returns: ReadableStream<Uint8Array>

close()

Close a stream when done writing.

await world.streams.close(runId, "default");

Parameters:

Parameter	Type	Description
`runId`	`string`	The workflow run ID
`name`	`string`	The stream name

list()

List all stream names associated with a workflow run.

const streamNames = await world.streams.list(runId);

Parameters:

Parameter	Type	Description
`runId`	`string`	The workflow run ID

Returns: string[]

getChunks()

Fetch stream chunks with cursor-based pagination. Unlike get() (which returns a live ReadableStream), this returns a snapshot of currently available chunks.

const result = await world.streams.getChunks(runId, "default", {
  limit: 50,
});
// result.data: StreamChunk[], result.cursor, result.hasMore, result.done

Parameters:

Parameter	Type	Description
`runId`	`string`	The workflow run ID
`name`	`string`	The stream name
`options.limit`	`number`	Max chunks per page (default: 100, max: 1000)
`options.cursor`	`string`	Cursor from a previous response

Returns: StreamChunksResponse

Field	Type	Description
`data`	`StreamChunk[]`	Chunks in index order. Each has `index` (0-based) and `data` (`Uint8Array`).
`cursor`	`string \| null`	Cursor for the next page
`hasMore`	`boolean`	Whether more pages of already-written chunks exist
`done`	`boolean`	Whether the stream is fully closed. When `false`, new chunks may appear in future requests even after `hasMore` is `false`.

getInfo()

Retrieve lightweight metadata about a stream without fetching chunks.

const info = await world.streams.getInfo(runId, "default");
// info.tailIndex: last chunk index (-1 if empty), info.done: whether stream is closed

Parameters:

Parameter	Type	Description
`runId`	`string`	The workflow run ID
`name`	`string`	The stream name

Returns: StreamInfoResponse

Field	Type	Description
`tailIndex`	`number`	Index of the last known chunk (0-based). `-1` when no chunks have been written.
`done`	`boolean`	Whether the stream is fully complete (closed).

Examples

Read a stream as a response

// app/api/workflow-streams/read/route.ts
import { getWorld } from "workflow/runtime";

export async function GET(req: Request) {
  const url = new URL(req.url);
  const streamName = url.searchParams.get("name") ?? "default";
  const runId = url.searchParams.get("runId")!;
  const world = await getWorld();
  const readable = await world.streams.get(runId, streamName);

  return new Response(readable, {
    headers: { "Content-Type": "application/octet-stream" },
  });
}

Paginate through stream chunks

import { getWorld } from "workflow/runtime";

const world = await getWorld();
let cursor: string | undefined;

do {
  const result = await world.streams.getChunks(runId, "default", { cursor });
  for (const chunk of result.data) {
    console.log(`Chunk ${chunk.index}:`, chunk.data);
  }
  cursor = result.cursor ?? undefined;
} while (cursor);

Streaming: Core concepts for streaming data from workflows
getWritable(): The standard way to write to streams from within steps
Storage: Query runs, steps, hooks, and events