Streaming Agent Methods in Go

A golem.AgentStream[T] is a sequence of values streamed between agents. It is an ordinary schema value: a method input field, a result, a struct field, an Option, a slice element, or the item of another stream.

package pipe
 
import "github.com/golemcloud/golem/sdks/go/golem"
 
type ID struct{ Name string }
 
type ValuesIn struct {
	Values golem.AgentStream[int32]
}
 
type Batch struct {
	Label  string
	Values golem.AgentStream[int32]
}
 
var Agent = golem.DefineAgent[ID](golem.Spec{Name: "PipeAgent"})
 
var (
	Sum     = Agent.Method[ValuesIn, int64]("sum")
	Doubled = Agent.Method[ValuesIn, golem.AgentStream[int32]]("doubled")
	Forward = Agent.Method[ValuesIn, golem.AgentStream[int32]]("forward")
	Batches = Agent.Method[golem.Unit, golem.AgentStream[Batch]]("batches")
)

A stream input is a field of the input struct, like any other parameter. golem.AgentStream[byte] is a byte stream; golem.AgentStream[golem.Binary] streams binary chunks.

Consume

agent.Handle(pipe.Sum, func(_ *golem.Context[state], in pipe.ValuesIn) int64 {
	var total int64
	for v := range in.Values.All() {
		total += int64(v)
	}
	return total
})

Next() returns (value, ok): ok == false is a clean end. Collect() reads everything and is only for bounded streams. Close() abandons a stream you stop reading; the producer sees it on its next write. Reading a stream that was closed or handed on, and an item that does not match the stream’s schema, panic: they are bugs, not conditions to handle.

Produce

agent.Handle(pipe.Doubled, func(_ *golem.Context[state], in pipe.ValuesIn) golem.AgentStream[int32] {
	return golem.ProduceStream(func(w *golem.AgentStreamWriter[int32]) error {
		for v := range in.Values.All() {
			if err := w.Write(v * 2); err != nil {
				return nil // the reader went away (golem.ErrReaderGone)
			}
		}
		return nil // clean end of stream
	})
})

• golem.ProduceStream runs the producer on its own goroutine and returns the reading end at once.
• Returning golem.ErrReaderGone, the error Write reports when the reader went away, ends the stream cleanly, so return err after a failed Write is fine. Any other error, or a panic, fails the invocation — the component traps, as in Rust — because a stream has no failure end and the reader must not see a clean end. Model recoverable errors as items, e.g. golem.AgentStream[golem.Result[T, E]].
• Stopping early does not close the input you were reading; call in.Values.Close() so its producer learns that nobody reads it.
• Write blocks until the consumer has room: that is the back-pressure, so nothing piles up in memory. Its only error is golem.ErrReaderGone; writing to a closed writer, or a value that cannot be encoded, panics.
• golem.NewAgentStream[T]() returns a writer and a stream for producing by hand; defer w.Close() ends it cleanly.
• golem.StreamOf(items...) is a stream of fixed values, handy for small results and tests.

Forward

Return or pass on an unread stream to move its endpoint; no items pass through this agent:

agent.Handle(pipe.Forward, func(_ *golem.Context[state], in pipe.ValuesIn) golem.AgentStream[int32] {
	return in.Values
})

A stream is affine: once returned, passed to a call or sent as an item, reading it again panics. A stream that was already read from cannot be forwarded: passing or returning it fails the call.

Ordering and Errors

• Items of one stream arrive in order. Sibling streams have no ordering between them; read them from separate goroutines when one producer could block on the other’s back-pressure.
• A bare stream has no error channel: a producer that returns an error fails the invocation rather than ending the stream, so the reader never mistakes a failure for a clean end. To let the reader see failures, model them as items: golem.AgentStream[golem.Result[T, E]].
• A decode failure panics in Next/All; it is never reported as a clean end.

Calling Streaming Methods

Same-component calls use the method descriptors as usual:

target := pipe.Agent.Get(pipe.ID{Name: "main"})
out := pipe.Doubled.MustCall(target, pipe.ValuesIn{Values: golem.StreamOf[int32](1, 2, 3)})
for v := range out.All() {
	fmt.Println(v)
}

CallAsync works too. Trigger and Schedule panic for a method that carries a stream anywhere in its input or output: both return before the call completes, so no stream could be handed over.

Guest clients that golem build generates for dependencies.agents (see golem-call-another-agent-go) include stream-bearing methods, spelling each stream as golem.AgentStream[T], including nested ones:

import provider "golem.local/bridge/stream-provider-guest-client"
 
remote := provider.MustGetStreamProvider(provider.StreamProviderID{Name: "main"})
total := remote.MustSum(golem.StreamOf[int32](1, 2, 3))
doubled := remote.MustDoubled(golem.ProduceStream(func(w *golem.AgentStreamWriter[int32]) error {
	return w.WriteAll(1, 2, 3)
}))

The external Go client generated for bridge.go.external (see golem-call-from-external-go) calls stream-bearing methods over an invocation session, spelling each stream bridge.AgentStream[T].

CLI and HTTP

golem agent invoke accepts - as the value of exactly one direct stream parameter to read it from stdin, one value per line (--stdin-format raw for a direct byte stream). Streamed results are printed as they arrive; --stdout-format raw writes a direct byte stream as bytes. Ctrl-C detaches without cancelling the durable invocation; --save-session and --resume-session reconnect.

Over HTTP, a stream-bearing endpoint is served through the Durable Streams protocol (see golem-durable-streams-go). MCP does not expose methods with streams; add a bounded, stream-free method for MCP callers.

Related Guides