golemcloud/golem/golem-streaming-agent-rust

Streaming Agent Methods in Rust

Use golem_rust::agentic::AgentStream<T> in an agent method signature. A stream is a recursive schema value: it may be a direct parameter or result, or appear inside a record, tuple, option, result, variant, list, or another stream. Do not assign streams static or preorder slot numbers; the runtime carries each endpoint with the value that contains it.

use golem_rust::agentic::{AgentStream, spawn_local};
 
#[derive(IntoSchema, FromSchema)]
pub struct Batch {
    pub label: String,
    pub values: AgentStream<u32>,
}
 
#[agent_definition]
pub trait StreamingAgent {
    fn new(name: String) -> Self;
    async fn sum(&self, values: AgentStream<u32>) -> u32;
    fn doubled(&self, values: AgentStream<u32>) -> AgentStream<u32>;
    fn batches(&self) -> (AgentStream<String>, AgentStream<Batch>);
}

Consume, Produce, and Forward

Read one item at a time with next().await. Ok(None) is normal close; Err is a decoding, transport, producer, or invocation failure, not EOF.

async fn sum_stream(mut values: AgentStream<u32>) -> Result<u32, String> {
    let mut total = 0;
    while let Some(value) = values.next().await? {
        total += value;
    }
    Ok(total)
}
 
fn doubled_stream(mut input: AgentStream<u32>) -> AgentStream<u32> {
    let (mut writer, output) = AgentStream::new();
    spawn_local(async move {
        loop {
            match input.next().await {
                Ok(Some(value)) => {
                    if writer.write_one(value * 2).await.is_err() {
                        break; // the reader dropped; stop producing
                    }
                }
                Ok(None) => break,
                Err(error) => panic!("Input stream failed: {error}"),
            }
        }
        // Dropping writer closes output normally.
    });
    output
}

write_one and each element of write_all wait for downstream acceptance, so a producer cannot run arbitrarily ahead. Acceptance is back-pressure, not proof that user code processed the item. Dropping the writer closes normally. Dropping the reader cancels that stream cooperatively; a later write reports that the reader was dropped.

Return an unread input stream directly to forward its endpoint without collecting or copying it:

fn forward(input: AgentStream<String>) -> AgentStream<String> {
    input
}

Streams are affine. Passing one into RPC, returning it, or forwarding it transfers ownership. Do not read, send, or clone aliases afterward. A partially consumed stream can be forwarded; the receiver reads its unread remainder. AgentStream::collect is convenient only for bounded streams; preserve streaming for large or unbounded data.

Ordering and Errors

• Items within one stream stay ordered.
• Independent sibling streams have no relative ordering guarantee. Drain them concurrently when

either producer could block on back-pressure.

• Bare agent streams have normal close but no recoverable terminal error payload. Use

AgentStream<Result<T, E>> for recoverable per-item or stream-local application errors.

• A producer panic, decode failure, transport failure, or fatal agent invocation failure fails the

active operation; it must not be treated as normal EOF.

• Dropping a stream is explicit consumer cancellation. It is distinct from an external transport

disconnect, which may only detach an observer while the durable invocation continues.

Calling Through Agent RPC

Definition clients and clients generated by golem build accept and return AgentStream values, including nested and sibling streams:

let target = StreamingAgentClient::get("main".to_string());
let (mut writer, input) = AgentStream::new();
spawn_local(async move {
    let _ = writer.write_all([1, 2, 3]).await;
});
 
let mut output = target.doubled(input).await;
while let Some(value) = output.next().await? {
    println!("{value}");
}

Generated guest clients provide schema-specific stream factories when a Rust type alone cannot identify the exact schema occurrence. Use those generated factories rather than constructing an ambiguous stream manually. Use awaited calls for stream-bearing methods. Definition-derived clients still expose trigger and scheduling helpers, but invoking them with streams is rejected at runtime (fallible paths return errors). CLI-generated guest clients omit these helpers for streaming methods. Stream-bearing constructors are rejected because client construction is synchronous.

For a native external application, declare bridge.rust.external in golem.yaml as described by the golem-call-from-external-rust guide. The generated client uses its exported affine AgentStream: create inputs with AgentStream::input, consume outputs with StreamExt, and call cancel() or drop an output when stopping early. A stream or call error reports producer, cancellation, transport, or invocation failure separately from clean end-of-stream.

CLI and MCP

The golem-invoke-agent-rust guide covers CLI invocation. - binds stdin to exactly one direct stream parameter; --stdin-format value|raw and --stdout-format value|raw control framing. Nested or multiple input streams require an SDK client. Structured output is a sequence of lifecycle documents, not one JSON object. Ctrl-C or disconnect detaches without cancelling the durable invocation; use --save-session and --resume-session to reconnect.

MCP uses Streamable HTTP as its transport, but it does not expose agent methods whose schemas contain agent streams. Keep an MCP-facing method stream-free or add a bounded adapter method.

Start and Verify

Start from the runnable Rust template, then build and invoke it:

golem new --yes --template rust/streaming streaming-agent
cd 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.