Skip to content

Run and stream agents

Genkit agents are built around conversations that continue across turns. A session or chat carries continuity, while each turn streams chunks and eventually resolves to a final output. This page covers starting a conversation, streaming a turn, and continuing from an earlier point.

The examples on this page use these imports:

import (
    "context"
    "encoding/json"
    "fmt"
    "time"


    "github.com/firebase/genkit/go/ai"
    aix "github.com/firebase/genkit/go/ai/exp"
    genkitx "github.com/firebase/genkit/go/genkit/exp"
)

Go calls an agent through the agent value itself. Run and RunText send one turn, Connect opens a live connection that carries many turns, and RunDetached starts a turn the server finishes on its own, returning a DetachedTask to wait on. Go has no Chat type: continuity is an invocation option on each call. See Background execution for detached work.

weatherAgent below is the *aix.Agent[WeatherState] that genkitx.DefineAgent returned; see Define agents.

Single-turn calls

Section titled “Single-turn calls”

Use RunText for the common text-only case. Use Run when you need to send a full aix.AgentInput, such as a resume payload or a detach request.

out, err := weatherAgent.RunText(ctx, "Weather in Tokyo?")
if err != nil {
    // The turn never started or could not produce a result; an in-band turn
    // failure instead resolves on out.FinishReason and out.Error.
    return fmt.Errorf("run turn: %w", err)
}


fmt.Println(out.Message.Text())
fmt.Println(out.SessionID)
fmt.Println(out.SnapshotID)

For a structured input:

out, err := weatherAgent.Run(ctx, &aix.AgentInput{
    Message: ai.NewUserTextMessage("Weather in Tokyo?"),
})

In-band failures resolve as an AgentOutput whose FinishReason is aix.AgentFinishReasonFailed, with structured details in out.Error. A non-nil Go error means the invocation did not start or could not produce an output. One case sets both: a run its caller stops, by cancelling the context or letting a deadline expire, returns the error that stopped it together with an output whose FinishReason is aix.AgentFinishReasonAborted and whose SnapshotID names where it stopped. Read out before giving up on err.

Turn input and output

Section titled “Turn input and output”

Three types carry everything that crosses the agent boundary, and each is small enough to read in full.

type AgentInput struct {
    // Detach moves the invocation to the background after this input.
    Detach bool `json:"detach,omitempty"`
    // Message is the user's input for this turn.
    Message *ai.Message `json:"message,omitempty"`
    // Resume answers an interrupted tool request instead of sending a new turn.
    Resume *ToolResume `json:"resume,omitempty"`
}

Run and RunText return *aix.AgentOutput[State], where State is the agent’s custom-state type. The type is generic, so a helper signature is *aix.AgentOutput[WeatherState], never a bare *aix.AgentOutput.

type AgentOutput[State any] struct {
    // Artifacts are the artifacts produced during the session.
    Artifacts []*Artifact `json:"artifacts,omitempty"`
    // Error is the structured failure, set when FinishReason is "failed" or "aborted".
    Error *status.Error `json:"error,omitempty"`
    // FinishReason is why the invocation finished.
    FinishReason AgentFinishReason `json:"finishReason,omitempty"`
    // Message is the last model response message of the conversation.
    Message *ai.Message `json:"message,omitempty"`
    // SessionID identifies the conversation. Stable across resumes.
    SessionID string `json:"sessionId,omitempty"`
    // SnapshotID is the most recent turn-end snapshot. Empty with no store.
    SnapshotID string `json:"snapshotId,omitempty"`
    // State is the final conversation state, for client-managed agents only.
    State *SessionState[State] `json:"state,omitempty"`
}

State is populated only when no session store is configured. A store-backed agent returns SnapshotID instead, and the state lives in the snapshot. AgentOutput carries no token or usage counts; read those from the trace.

type AgentStreamChunk struct {
    // Artifact is a newly produced artifact.
    Artifact *Artifact `json:"artifact,omitempty"`
    // CustomPatch is an RFC 6902 JSON Patch against the custom state document.
    CustomPatch JSONPatch `json:"customPatch,omitempty"`
    // ModelChunk holds generation tokens from the model.
    ModelChunk *ai.ModelResponseChunk `json:"modelChunk,omitempty"`
    // TurnEnd is non-nil once the agent finishes the current input.
    TurnEnd *TurnEnd `json:"turnEnd,omitempty"`
}


type TurnEnd struct {
    // FinishReason is why this turn finished.
    FinishReason AgentFinishReason `json:"finishReason,omitempty"`
    // SnapshotID is the snapshot persisted at the end of this turn, if any.
    SnapshotID string `json:"snapshotId,omitempty"`
}

Those four fields are the whole chunk, and more than one can be set on a single chunk. There is no interrupt field and no detach field: interrupts arrive on chunk.ModelChunk, so read them with chunk.ModelChunk.Interrupts(), and a detach is reported on AgentOutput.FinishReason. Tool requests and responses stream as ordinary model chunk content, so a tool-call indicator reads chunk.ModelChunk.Content.

Finish reasons

Section titled “Finish reasons”

The first six values are forwarded verbatim from the model’s own finish reason. The last three are agent-specific and never arise from a model.

ConstantWire valueMeaning
aix.AgentFinishReasonStopstopThe model stopped naturally.
aix.AgentFinishReasonLengthlengthGeneration hit the token limit.
aix.AgentFinishReasonBlockedblockedGeneration was blocked, usually by a safety filter.
aix.AgentFinishReasonInterruptedinterruptedA tool paused for input. See Agent interrupts.
aix.AgentFinishReasonOtherotherThe model stopped for some other reason.
aix.AgentFinishReasonUnknownunknownThe model gave no reason.
aix.AgentFinishReasonAbortedabortedThe caller stopped the run: a cancelled context, an expired deadline, a closed transport, a limit it set such as ai.WithMaxTurns, or Abort on a detached run. The snapshot keeps the turns that finished.
aix.AgentFinishReasonDetacheddetachedThe client detached and the work continues in the background.
aix.AgentFinishReasonFailedfailedA turn broke. Read out.Error. The snapshot keeps the tool rounds the turn completed.

Invocation options

Section titled “Invocation options”

Run, RunText, and Connect all take the same aix.InvocationOption[State] values.

func WithSessionID[State any](id string) InvocationOption[State]
func WithSnapshotID[State any](id string) InvocationOption[State]
func WithState[State any](state *SessionState[State]) InvocationOption[State]
  • aix.WithSessionID[State](id) resumes the latest server-managed snapshot for a conversation.
  • aix.WithSnapshotID[State](id) resumes or branches from a specific server-managed snapshot. See Session stores.
  • aix.WithState[State](state) continues a client-managed conversation by sending the full state.

WithState is mutually exclusive with WithSessionID and WithSnapshotID. WithSessionID and WithSnapshotID can be combined to assert that the snapshot belongs to the session.

next, err := weatherAgent.RunText(ctx, "What about Paris?",
    aix.WithSessionID[WeatherState](out.SessionID),
)

Because all three return the same interface type, you can build the list up and spread it into any entry point:

opts := []aix.InvocationOption[WeatherState]{}
if sessionID != "" {
    opts = append(opts, aix.WithSessionID[WeatherState](sessionID))
}


out, err := weatherAgent.RunText(ctx, "What about Paris?", opts...)

Bounding a turn

Section titled “Bounding a turn”

There is no per-turn deadline option. Cap the tool loop inside a turn with ai.WithMaxTurns(n) in the agent’s aix.InlinePrompt, and bound wall-clock time with context.WithTimeout on the context you pass to Run, RunText, or Connect.

ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()


out, err := weatherAgent.RunText(ctx, "Plan a two-week itinerary.")

Cancelling that context stops the invocation. Run and RunText return the cancellation error together with an output whose FinishReason is aix.AgentFinishReasonAborted. The turn that was in flight is discarded whole, and the aborted snapshot the output names holds the turns that finished, so it is a resume point. ai.WithMaxTurns ends a turn the same way, because a limit the caller set is a caller stop rather than a failure. A custom agent should still check ctx.Err() between turns.

Re-attempt a turn

Section titled “Re-attempt a turn”

An input with neither Message nor Resume runs the last turn again on the conversation as it stands. That is how a failed or aborted snapshot is picked up without repeating the tool calls that already succeeded:

retried, err := weatherAgent.Run(ctx, &aix.AgentInput{},
    aix.WithSnapshotID[WeatherState](out.SnapshotID),
)

An empty input is rejected with INVALID_ARGUMENT only when the session has no messages to continue. A new message on the same snapshot changes course instead, and the previous snapshot ID rewinds past the turn altogether. See Agent error handling for deciding whether a retry is worth making.

Open a bidirectional stream

Section titled “Open a bidirectional stream”

Use Connect for multi-turn local clients and streaming UIs. The connection lets you send text, messages, resume payloads, or a detach signal while receiving chunks.

Reach for Connect when the caller needs direct control over both sides of the conversation on one live connection. It is useful for command-line tools, local services, workers, and lower-level integrations that need to stream output, observe custom state patches, handle interrupts, send a resume payload, or send another message after a TurnEnd without reconnecting.

For most single-turn server code, use RunText or Run. For browser, mobile, and other HTTP clients, serve the agent over HTTP and drive it from the prebuilt client rather than managing a bidirectional stream directly.

Connect takes the same invocation options as Run and RunText, so a streaming connection can resume a stored conversation:

conn, err := weatherAgent.Connect(ctx, aix.WithSessionID[WeatherState](previousSessionID))

A full turn over a fresh connection:

conn, err := weatherAgent.Connect(ctx)
if err != nil {
    // Connect fails when the init payload is rejected before any turn runs.
    return fmt.Errorf("connect to agent: %w", err)
}
defer conn.Close()


if err := conn.SendText("Weather in Tokyo?"); err != nil {
    return fmt.Errorf("send message: %w", err)
}


for chunk, err := range conn.Receive() {
    if err != nil {
        // A stream error ends the turn, such as the context being cancelled.
        return fmt.Errorf("stream turn: %w", err)
    }
    if chunk.ModelChunk != nil {
        fmt.Print(chunk.ModelChunk.Text())
    }
    if chunk.TurnEnd != nil {
        fmt.Printf("\nturn finished: %s\n", chunk.TurnEnd.FinishReason)
        break
    }
}


out, err := conn.Output()
if err != nil {
    return fmt.Errorf("finalize turn: %w", err)
}
fmt.Println(out.SnapshotID)

Breaking from Receive does not cancel the connection. Multi-turn clients commonly break on TurnEnd, send another input, and call Receive again.

The cli.go file of go/samples/basic-agents is a complete client written this way: it streams each turn, renders tool calls, routes interrupts, and drives detach and resume, all against the Agent and AgentConnection surface.

Connection lifecycle

Section titled “Connection lifecycle”
  • conn.Close() signals that no more inputs will be sent. Write defer conn.Close() right after Connect so an early return on an error path still releases the invocation.
  • conn.Output() is the terminator. It closes the input side for you, drains any chunks Receive did not consume, and blocks until the agent finalizes. It is idempotent, so the deferred Close and a later Output() do not conflict.
  • conn.Done() returns a channel closed when the invocation completes, for a caller that waits on it in a select.

Concurrency

Section titled “Concurrency”

An Agent value is immutable after definition and safe for concurrent use. Share one *aix.Agent[State] across every HTTP handler and call Run, RunText, Connect, RunDetached, GetSnapshot, GetLatestSnapshot, WaitForSnapshot, and Abort from any goroutine. A DetachedTask holds a snapshot ID and nothing else, so it is safe to share too.

An AgentConnection belongs to one invocation and is not a shared object. Do not call Output() from one goroutine while another iterates Receive(): both consume the stream and would split chunks between them. Finish Receive first.

Live custom state

Section titled “Live custom state”

AgentConnection applies streamed custom-state patches as it receives chunks. Read conn.Custom() to inspect the custom state observed so far.

for chunk, err := range conn.Receive() {
    if err != nil {
        return fmt.Errorf("stream turn: %w", err)
    }
    if len(chunk.CustomPatch) > 0 {
        state, err := conn.Custom()
        if err != nil {
            // Fails if an applied patch cannot decode into the State type.
            return fmt.Errorf("read custom state: %w", err)
        }
        renderState(state)
    }
}

Custom() returns (State, error), the state value itself rather than a pointer, so there is nothing to nil-check. Before the first patch of a turn arrives it returns the zero value of State. The error is non-nil only when an applied patch cannot decode into State. The patch itself is an RFC 6902 JSON Patch rooted at the custom document; see Sessions and state.

The authoritative final state is on AgentOutput.State for client-managed agents, or in the saved snapshot for server-managed agents.

Agent handles

Section titled “Agent handles”

Code that knows an agent only by name, such as an orchestrator, a middleware, or a tool, drives it through an *aix.AgentHandle: the same agent with its custom state fixed to json.RawMessage. genkitx.LookupAgent finds one in the registry, and agent.Handle() returns one for an agent value you already hold.

h := genkitx.LookupAgent(g, "weather") // nil on a miss, like every Lookup
if h == nil {
    return fmt.Errorf("no agent named %q", "weather")
}


out, err := h.RunText(ctx, "Weather in Tokyo?",
    aix.WithSessionID[json.RawMessage](sessionID),
)

A handle has every call the typed agent has (Run, RunText, RunDetached, Task, GetSnapshot, GetLatestSnapshot, WaitForSnapshot, and Abort), plus Name() and Metadata(), which reports whether the agent is server-managed and abortable. Its invocation options are the same aix.InvocationOption values typed at json.RawMessage, resolved through the same code as the typed calls, so it rejects the same inputs with the same wording. Every read goes through the agent’s companion actions: the state transform applies and a stale detached row reads as expired, exactly as over HTTP. LookupAgent needs no genkit.WithExperimental(), since it only reads the registry and only the gated constructors can register an agent.

Next steps

Section titled “Next steps”