> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agenticenv.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Streaming
> Stream live tokens and lifecycle events from an agent run
Use streaming when your application needs live output — chat UIs, progress indicators, or AG-UI-compatible frontends.
[`Agent.Stream`](https://pkg.go.dev/github.com/agenticenv/agent-sdk-go/pkg/agent#Agent.Stream) returns an [`AgentStream`](https://pkg.go.dev/github.com/agenticenv/agent-sdk-go/pkg/agent#AgentStream) handle. Call `Events` for the live channel. For a final result without live tokens, use [Run](/getting-started/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`](/examples/reconnect) to resume after reconnect |
```go theme={null}
// 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`](/getting-started/run#agentrun-methods). 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`](/getting-started/llm-providers) reports [`IsStreamSupported()`](https://pkg.go.dev/github.com/agenticenv/agent-sdk-go/pkg/interfaces#LLMClient) 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.
```go theme={null}
a, err := agent.NewAgent(
agent.WithLLMClient(llmClient),
agent.WithSystemPrompt("You are a helpful assistant."),
)
```
Then call `Stream`:
```go theme={null}
// 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`](https://pkg.go.dev/github.com/agenticenv/agent-sdk-go/pkg/agent#AgentStreamOptions):
```go theme={null}
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`](https://pkg.go.dev/github.com/agenticenv/agent-sdk-go/pkg/agent#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, delegation, and budget)
| Event | When |
| - | - |
| `AgentEventTypeCustom` | Tool approval, sub-agent delegation, or `budget_approval` during `Stream` |
Parse approval events with `ParseCustomEventApproval` / `ParseCustomEventDelegation` / `ParseCustomEventBudget`, then call [`AgentStream.Approve`](/features/approvals) with the `ApprovalToken`. See [Approvals](/features/approvals) and [Budget](/features/budget).
## 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`](/features/token-usage):
```go theme={null}
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](/features/token-usage) for field details.
## Reconnect with GetAgentStream
On a durable runtime ([Temporal](/runtimes/temporal) or [Restate](/runtimes/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:
```go theme={null}
// 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](/advanced/timeouts-and-modes#what-each-context-does).
`LocalRuntime` does not support crash reconnect — non-zero offsets return `ErrStreamOffsetNotSupported`. Full protocol: [Durable Execution](/advanced/durable-execution#client-side-stream-recovery). Runnable demo: [Reconnect](/examples/reconnect).
## Streaming with conversation history
Pass `ConversationOptions` on [`AgentStreamOptions`](https://pkg.go.dev/github.com/agenticenv/agent-sdk-go/pkg/agent#AgentStreamOptions) to share history across turns while streaming:
```go theme={null}
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](/features/conversation) for backend options and lifecycle management.
## AG-UI protocol
Stream events follow the [AG-UI open protocol](https://docs.ag-ui.com). Serialize any event with `event.ToJSON()` and forward over SSE or WebSocket to a compatible frontend. See [AG-UI Protocol](/features/ag-ui-protocol).
## Examples
Partial tokens, tool events, and RUN\_FINISHED handling
Resume a stream from a saved offset after a crash
SSE server + CopilotKit UI over AG-UI events
AgentRun — Get, Done, Status, Cancel
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.