Streaming

Use streaming when your application needs live output — chat UIs, progress indicators, or AG-UI-compatible frontends. Agent.Stream returns an AgentStream handle. Call Events for the live channel. For a final result without live tokens, use Run instead.

AgentStream methods

Method	Purpose
`ID()`	Stable run id — available immediately; persist before consuming events for crash-recovery on Temporal
`Status(ctx)`	Current lifecycle status (`running`, `completed`, `failed`, `cancelled`, …)
`Cancel(ctx)`	Request cancellation of the agent run (same effect as cancelling the `Stream` context)
`Done()`	Channel closed when the run finishes (success or failure)
`Get(ctx)`	Block until finished; return the result. Cancelling `ctx` only unblocks `Get` — use `Cancel` or cancel `Stream`’s ctx to stop the run
`Events(ctx, …)`	Event channel; cancelling this `ctx` stops the subscriber only (Temporal). Optional `WithOffset` to resume after reconnect

// Cancelling streamCtx cancels the agent run. Use a separate eventsCtx if you need
// to drop the subscriber without stopping the run (reconnect / crash simulation).
streamCtx := ctx
agentStream, err := a.Stream(streamCtx, prompt, nil)
if err != nil {
    return err
}
_ = agentStream.ID()

st, _ := agentStream.Status(ctx)
fmt.Println("status:", st)

eventCh, err := agentStream.Events(ctx)
if err != nil {
    return err
}
for ev := range eventCh {
    // handle events
}

// Optional: wait for completion / read result after the channel closes
<-agentStream.Done()
result, err := agentStream.Get(ctx)

Status / Cancel / Done / Get match AgentRun. Streaming adds Events.

Enable streaming

Create an agent, then call Stream. Stream enables LLM token deltas and delivers AG-UI lifecycle events: LLM token deltas (TEXT_MESSAGE_CONTENT) are emitted only when the underlying LLMClient reports IsStreamSupported() as true (OpenAI, Anthropic, Gemini, DeepSeek, and Ollama do). If the provider does not support streaming, you still get lifecycle and tool events — the assistant text arrives as a complete message rather than partial deltas.

a, err := agent.NewAgent(
    agent.WithLLMClient(llmClient),
    agent.WithSystemPrompt("You are a helpful assistant."),
)

Then call Stream:

// runID is available synchronously before any event arrives.
// Persist it alongside your correlation key (conversationID, sessionID) before
// consuming events — this is what GetAgentStream needs after a process crash.
agentStream, err := a.Stream(ctx, "What's 17 * 23?", nil)
if err != nil {
    return err
}
runID := agentStream.ID()
_ = runID // store runID in your DB/cache here

eventCh, err := agentStream.Events(ctx)
if err != nil {
    return err
}

for ev := range eventCh {
    if ev == nil {
        continue
    }
    switch ev.Type() {
    case agent.AgentEventTypeTextMessageContent:
        if t, ok := ev.(*agent.AgentTextMessageContentEvent); ok {
            fmt.Print(t.Delta)
        }
    case agent.AgentEventTypeToolCallStart:
        if t, ok := ev.(*agent.AgentToolCallStartEvent); ok {
            fmt.Printf("\n[tool] %s\n", t.ToolCallName)
        }
    case agent.AgentEventTypeRunFinished:
        fmt.Println("\n[done]")
    case agent.AgentEventTypeRunError:
        if t, ok := ev.(*agent.AgentRunErrorEvent); ok {
            fmt.Printf("\n[error] %s\n", t.Message)
        }
    }
}

Always check if ev == nil { continue } before calling ev.Type(). The channel can deliver a nil sentinel to signal the end of a batch.

Event guarantees

On all runtimes, the event channel is guaranteed to close only after a terminal RUN_FINISHED or RUN_ERROR event is delivered when the run itself ends. This holds even on Temporal if:

Your process crashes during streaming
The stream connection fails transiently
You cancel the Events context (subscriber disconnect) — the agent run continues; reconnect with GetAgentStream

Cancelling the Stream context cancels the agent run and surfaces a terminal event on active subscribers. If you reconnect via GetAgentStream with a saved offset, you will not miss the terminal event (offsets may redeliver — discard duplicates at or below the saved offset). For in-process streaming, a stream error (rare) will still deliver a terminal event before channel close.

When using Temporal for crash-recovery, always save both the run ID and offset before processing each event. If your process crashes between reading an event and saving its offset, only that one event will replay on reconnect.

Disable LLM token streaming

By default, Stream enables LLM token streaming — the model emits partial text as TEXT_MESSAGE_CONTENT deltas. Lifecycle events (tools, approvals, RUN_FINISHED, …) still stream either way. To turn off token deltas and receive a single complete assistant message instead, set DisableTokenStreaming on AgentStreamOptions:

agentStream, err := a.Stream(ctx, prompt, &agent.AgentStreamOptions{
    DisableTokenStreaming: true,
})
if err != nil {
    return err
}
eventCh, err := agentStream.Events(ctx)

Use this when you want AG-UI / tool lifecycle events without partial token delivery (for example, lower provider stream overhead or a UI that only renders the final message).

Event types

Stream events are typed AgentEvent values. Use ev.Type() and type-assert to the concrete struct to access fields.

Lifecycle

Event	When
`AgentEventTypeRunStarted`	Run begins
`AgentEventTypeRunFinished`	Run completes — carries final result on `AgentRunFinishedEvent.Result`
`AgentEventTypeRunError`	Run failed — error message on `AgentRunErrorEvent.Message`
`AgentEventTypeStepStarted`	Sub-agent child workflow begins — `StepName` is the sub-agent route name
`AgentEventTypeStepFinished`	Sub-agent child workflow ends

Text output

Event	When
`AgentEventTypeTextMessageStart`	Assistant message begins
`AgentEventTypeTextMessageContent`	Partial text delta — print `Delta` as it arrives
`AgentEventTypeTextMessageEnd`	Assistant message ends

Tool calls

Event	When
`AgentEventTypeToolCallStart`	Tool invocation begins — `ToolCallName` identifies the tool
`AgentEventTypeToolCallArgs`	Partial tool arguments streaming in
`AgentEventTypeToolCallEnd`	Tool invocation ends
`AgentEventTypeToolCallResult`	Tool result content

Reasoning

Event	When
`AgentEventTypeReasoningMessageContent`	Partial reasoning/thinking delta (Anthropic when extended thinking is enabled)

Custom (approvals and delegation)

Event	When
`AgentEventTypeCustom`	Approval or delegation requests during `Stream`

Parse approval events with ParseCustomEventApproval / ParseCustomEventDelegation, then call AgentStream.Approve with the ApprovalToken. See Approvals for the full flow.

Displaying streamed text

TEXT_MESSAGE_CONTENT deltas and AgentRunFinishedEvent.Result.Content carry the same text. Do not print both. If you rendered deltas live, skip printing result.Content from RUN_FINISHED.

When sub-agents are configured, events from delegated runs fan in to the parent stream. Use AgentName on typed events to identify which agent emitted them. Multiple RUN_FINISHED events may appear before the root run finishes — each corresponds to one completed sub-agent run.

Token usage from Stream

Aggregated token counts are on Result.LLMUsage inside AgentRunFinishedEvent:

for ev := range eventCh {
    if ev == nil {
        continue
    }
    if ev.Type() == agent.AgentEventTypeRunFinished {
        if fe, ok := ev.(*agent.AgentRunFinishedEvent); ok && fe.Result != nil {
            fmt.Printf("Tokens: %d prompt, %d completion\n",
                fe.Result.LLMUsage.PromptTokens,
                fe.Result.LLMUsage.CompletionTokens,
            )
        }
    }
}

See Token Usage for field details.

Reconnect with GetAgentStream

On a durable runtime (Temporal or Restate), persist agentStream.ID() and each event’s offset before handling the event. After a process crash, reconnect and resume — the run keeps executing server-side:

// During initial stream
for ev := range eventCh {
    if ob, ok := ev.(interface{ Offset() (int64, bool) }); ok {
        if off, has := ob.Offset(); has {
            db.SaveOffset(runID, off) // before handling the event
        }
    }
    handleEvent(ev)
}

// After crash/restart
savedRunID, savedOffset := db.LoadSavedRun()
s, err := a.GetAgentStream(ctx, savedRunID)
if err != nil {
    if errors.Is(err, agent.ErrRunAlreadyCompleted) {
        // Workflow finished while disconnected — load outcome from
        // conversation/memory, or start a new run.
        return nil
    }
    return err // including ErrStreamNotFound
}
eventCh, err := s.Events(ctx, agent.WithOffset(savedOffset))
// Continue consuming from exactly where you left off

The offset is opaque and is attached only to events that support reconnect (lifecycle, text, tools, custom events all carry it). Cancelling GetAgentStream or Events does not stop the run — use AgentStream.Cancel for that. After reconnect, WithTimeout (if set) starts fresh. Full cancel/timeout rules: Timeouts & Modes. LocalRuntime does not support crash reconnect — non-zero offsets return ErrStreamOffsetNotSupported. Full protocol: Durable Execution. Runnable demo: Reconnect.

Streaming with conversation history

Pass ConversationOptions on AgentStreamOptions to share history across turns while streaming:

conv := inmem.NewConversation(inmem.WithMaxSize(100))

a, err := agent.NewAgent(
    agent.WithLLMClient(llmClient),
    agent.WithConversation(conversation.Config{Conversation: conv, Size: 20}),
)

opts := &agent.AgentStreamOptions{
    ConversationOptions: &agent.ConversationOptions{ID: "session-1"},
}

// First turn
agentStream, _ := a.Stream(ctx, "My name is Alice.", opts)
eventCh, _ := agentStream.Events(ctx)
for ev := range eventCh { /* handle */ }

// Second turn — agent remembers the session
agentStream, _ = a.Stream(ctx, "What's my name?", opts)
eventCh, _ = agentStream.Events(ctx)
for ev := range eventCh { /* handle */ }

See Conversation for backend options and lifecycle management.

AG-UI protocol

Stream events follow the AG-UI open protocol. Serialize any event with event.ToJSON() and forward over SSE or WebSocket to a compatible frontend. See AG-UI Protocol.

Examples

Stream example

Partial tokens, tool events, and RUN_FINISHED handling

Reconnect

Resume a stream from a saved offset after a crash

AG-UI

SSE server + CopilotKit UI over AG-UI events

Run

AgentRun — Get, Done, Status, Cancel

Overview

Getting Started

Runtimes

Features

Advanced

Observability

Testing & Evaluation

Production

Reference Apps

AgentStream methods

Enable streaming

Event guarantees

Disable LLM token streaming

Event types

Lifecycle

Text output

Tool calls

Reasoning

Custom (approvals and delegation)

Displaying streamed text

Token usage from Stream

Reconnect with GetAgentStream

Streaming with conversation history

AG-UI protocol

Examples

Stream example

Reconnect

AG-UI

Run

​AgentStream methods

​Enable streaming

​Event guarantees

​Disable LLM token streaming

​Event types

​Lifecycle

​Text output

​Tool calls

​Reasoning

​Custom (approvals and delegation)

​Displaying streamed text

​Token usage from Stream

​Reconnect with GetAgentStream

​Streaming with conversation history

​AG-UI protocol

​Examples

Stream example

Reconnect

AG-UI

Run

AgentStream methods

Enable streaming

Event guarantees

Disable LLM token streaming

Event types

Lifecycle

Text output

Tool calls

Reasoning

Custom (approvals and delegation)

Displaying streamed text

Token usage from Stream

Reconnect with GetAgentStream

Streaming with conversation history

AG-UI protocol

Examples