Streaming Agent Methods in TypeScript

Declare a stream with s.stream(itemSchema) and use AgentStream<T> in the implementation. Streams are recursive schema values: they may be direct inputs or outputs, nested in records, tuples, options, results, variants, or lists, and even occur inside streamed items. Do not model them as static or preorder stream slots.

import { AgentStream, defineAgent, method, s } from '@golemcloud/golem-ts-sdk';
import { z } from 'zod';
 
const U32Stream = s.stream(s.u32()) as unknown as z.ZodType<
  AgentStream<number>,
  AgentStream<number>
>;
const StringStream = s.stream(z.string()) as unknown as z.ZodType<
  AgentStream<string>,
  AgentStream<string>
>;
const Batch = z.object({
  label: z.string(),
  values: U32Stream,
});
const BatchStream = s.stream(Batch) as unknown as z.ZodType<
  AgentStream<z.infer<typeof Batch>>,
  AgentStream<z.infer<typeof Batch>>
>;
 
const StreamingAgent = defineAgent({
  name: 'StreamingAgent',
  id: { name: z.string() },
  methods: {
    sum: method({ input: { values: s.stream(s.u32()) }, returns: s.u32() }),
    doubled: method({
      input: { values: s.stream(s.u32()) },
      returns: s.stream(s.u32()),
    }),
    siblings: method({
      input: {},
      returns: z.tuple([StringStream, BatchStream]),
    }),
  },
});

Consume, Produce, and Forward

AgentStream is an async iterator. next() returns { done: true } only for normal close.

export const StreamingAgentImpl = StreamingAgent.implement({
  init: () => ({}),
  methods: {
    async sum({ values }) {
      let total = 0;
      for await (const value of values) total += value;
      return total;
    },
    doubled({ values }) {
      return AgentStream.from((async function* () {
        for await (const value of values) yield value * 2;
      })());
    },
  },
});

AgentStream.from is lazy. Across RPC, each accepted write gates the next source pull and provides back-pressure. An early break from for await calls return() and closes the readable endpoint. Call await stream.return() yourself when abandoning a stream outside a loop. throw(reason) closes locally and rejects with that local reason; the reason is not sent to the producer.

Forward an unread input by returning the same object. Encoding, passing, returning, or forwarding a stream transfers ownership; do not reuse it afterward. A stream is single-reader and operations must not overlap. Do not collect a potentially large stream into an array merely to cross RPC.

Ordering and Errors

• One stream preserves item order. Independent sibling streams have no cross-stream ordering guarantee; consume them concurrently if one could block the other through back-pressure.
• Clean EOF, explicit return()/drop, an aborted call, transport disconnect, and fatal invocation failure are different outcomes.
• P3 agent streams do not carry a recoverable terminal error. Model recoverable failures as items, for example s.stream(s.result(valueSchema, errorSchema)).
• A source iterator or cleanup rejection fails the active production or invocation session rather than becoming clean EOF. Reader drop is cooperative and cannot interrupt an arbitrary pending source next().

Calling Through Agent RPC

Same-component definition clients and cross-component clients generated by golem build use the same AgentStream values:

const target = StreamingAgent.client.get({ name: 'main' });
const output = await target.doubled({ values: AgentStream.from([1, 2, 3]) });
 
for await (const value of output) {
  console.log(value);
}

For cross-component calls, declare dependencies.agents and follow the golem-call-another-agent-ts guide. Generated guest clients preserve nested stream locations. Stream-bearing methods only support awaited calls: .trigger, .schedule, and .scheduleCancelable are omitted. Stream-bearing constructors are not supported.

For external Node.js clients, declare bridge.ts.external as described by golem-call-from-external-ts. The generated package exports single-consumer AgentStream<T> and the agentStream(asyncIterable) input wrapper. Iterate returned streams once; breaking iteration closes the output. Streaming calls also expose .abortable(signal, ...args). Aborting or dropping the transport is cancellation, not normal EOF, and remote side effects may already have occurred.

CLI and MCP

The golem-invoke-agent-ts guide covers CLI streams. - binds stdin to exactly one direct stream parameter. Value mode reads one item per line; raw mode is only for one direct stream<binary> or stream<u8>. Nested and multiple inputs require an SDK client. Structured results are lifecycle documents. Ctrl-C or disconnect detaches without cancelling the durable invocation; save and resume the CLI session when reconnection is required.

MCP’s Streamable HTTP transport is unrelated to agent stream values. MCP currently refuses any agent method whose input or output schema contains a stream. Expose a bounded stream-free adapter method when the agent must also be callable through MCP.

Start and Verify

golem new --yes --language ts streaming-agent
golem build
golem deploy --yes

For complete input, output, mixed, nested, sibling, forwarding, cancellation, and failure examples, see the repository’s cross-language streaming RPC fixture .