Read, write, and manage real-time data streams for workflow runs.
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 { getWorld } from "workflow/runtime";
const world = await getWorld();
// Stream methods are called on world.streams — e.g. world.streams.write()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.doneParameters:
| 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 closedParameters:
| 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). |
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