Run and stream agents

The JavaScript client exposes a high-level interface for driving an agent across turns.

• AgentApi is the agent handle that remoteAgent() returns. You call chat(), loadChat(), getSnapshot(), and abort() on it; the per-turn methods live on the AgentChat that chat() returns.
• AgentChat is a stateful conversation. It sends normal turns, streams turns, resumes interrupts, detaches work, and tracks the next-turn state, including the session ID, for you.
• AgentTurn represents one in-flight streaming turn. It gives you a stream, a final response, and an abort helper.
• AgentResponse is the completed turn, with text, tool requests, interrupts, finish reason, snapshot ID, custom state, artifacts, and raw output.
• AgentChunk is one streamed update. It can contain text, accumulated text, model data, tool requests, custom state, or an artifact.
• AgentInterrupt is a paused tool request. It has the original input and helpers for building resume payloads.
• DetachedTask is a background task handle. It can poll, wait, or abort the detached turn.

res.state and chat.state are shortcuts for the custom state. Use res.raw.state when you need the full session state with messages, artifacts, and custom state together.

A local agent from ai.defineAgent() and a remote client from remoteAgent() share this interface, so the same code drives both.

const chat = weatherAgent.chat();
const res = await chat.send('Weather in Tokyo?');

console.log(res.text);
console.log(res.sessionId);
console.log(res.snapshotId);
console.log(res.state);

Calling chat() without arguments starts a new conversation. Pass sessionId for the latest server-managed conversation, snapshotId when you need an exact saved point, or state when the client owns the full session state.

const chat = weatherAgent.chat({
  sessionId: 'user-session-123',
});

await chat.send('What did we discuss last time?');

When both sessionId and snapshotId are supplied, the snapshot selects the exact resume point and the session ID acts as an ownership guard.

loadChat() reads a server snapshot and hydrates messages, custom state, artifacts, snapshotId, and sessionId before the next turn.

const chat = await weatherAgent.loadChat({ sessionId: 'user-session-123' });

console.log(chat.messages.length);
console.log(chat.state);

await chat.send('Continue from there.');

Use getSnapshot() when you only need to inspect a snapshot, such as a status page or audit view. Use loadChat() when you want to continue the conversation from that saved state.

const chat = weatherAgent.chat();
const turn = chat.sendStream('Weather in Tokyo?');

for await (const chunk of turn.stream) {
  if (chunk.text) process.stdout.write(chunk.text);
  if (chunk.custom) updateStatus(chunk.custom);
  if (chunk.artifact) renderArtifact(chunk.artifact);
}

const res = await turn.response;
console.log(res.finishReason);

The non-streaming send() path drains the stream internally so custom state patches are still applied. This keeps send() and sendStream() consistent for server-managed agents, where final wire output may return a snapshotId instead of full state.

Cancel a foreground turn from the caller.

const controller = new AbortController();
const turn = chat.sendStream('Write a long report.', {
  abortSignal: controller.signal,
});

setTimeout(() => controller.abort(), 1000);

const res = await turn.response;
console.log(res.finishReason);

You can also call turn.abort(). Foreground aborts return an aborted response when cancellation is observed.

When a turn fails after the invocation starts, the client throws AgentError. The error carries the last-good state, snapshot ID, and response object when available.

import { AgentError } from 'genkit/beta/client';

try {
  await chat.send('Use a broken tool.');
} catch (err) {
  if (err instanceof AgentError) {
    console.error(err.status);
    console.error(err.snapshotId);
    console.error(err.state);
  }
}

Initialization misuse, such as sending state to a server-managed agent or sessionId to a client-managed agent, is rejected before a turn starts.

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.

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.

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.

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.

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...)

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.

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.

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.

• 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.

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.

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.

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.

• Sessions and state covers custom state, artifacts, and how patches are produced.
• Session stores covers snapshot IDs, branching, and store implementations.
• Agent interrupts covers building the Resume payload.
• Background execution covers detached tasks, waiting, and abort.
• Serve agents over HTTP covers the wire protocol and browser clients.
• Agent error handling covers failed turns and rejected init payloads.

Local agents from ai.defineAgent() and remote clients from remoteAgent() share this interface, so the same code drives both.

final chat = weatherAgent.chat();
final res = await chat.send(text: 'Weather in Tokyo?');

print(res.text);
print(res.snapshotId);
print(res.state);

Calling chat() without arguments starts a new conversation. Pass sessionId to resume or start a server-managed conversation under that session ID, snapshotId when you need to resume from an exact snapshot, or state to seed or carry forward a client-managed state.

final chat = weatherAgent.chat(
  sessionId: 'user-session-123',
);

await chat.send(text: 'What did we discuss last time?');

loadChat() reads a server snapshot and hydrates messages, custom state, artifacts, snapshotId, and sessionId before the next turn.

final chat = await weatherAgent.loadChat(sessionId: 'user-session-123');

print(chat.messages.length);
print(chat.state);

await chat.send(text: 'Continue from there.');

Use getSnapshot() when you only need to inspect a snapshot, such as for an audit view. Use loadChat() when you want to continue the conversation from that saved state.

final chat = weatherAgent.chat();
final turn = chat.sendStream(text: 'Weather in Tokyo?');

await for (final chunk in turn.stream) {
  if (chunk.text.isNotEmpty) stdout.write(chunk.text);
  if (chunk.custom != null) updateStatus(chunk.custom!);
  if (chunk.artifact != null) renderArtifact(chunk.artifact!);
}

final res = await turn.response;
print(res.finishReason.value);

The non-streaming send() path drains the stream internally so custom state patches are still applied. This keeps send() and sendStream() consistent for server-managed agents.

Every turn method (send, sendStream, and detach) accepts an optional context map. Use it to pass ambient request data, such as auth, that tools and custom agents can read without exposing it to the model.

final res = await chat.send(
  text: 'What is on my calendar today?',
  context: {
    'auth': {'name': 'Ada'},
  },
);

Tools read the context through the tool context (ctx.context), and custom agents read it through options.context.

Per-turn context is honored by the in-process transport, where you drive a local agent from ai.defineAgent(). A remoteAgent() over HTTP rejects a non-empty context with an UnsupportedError, because a remote agent derives its context server-side from the incoming request.

Cancel a foreground turn using turn.abort() on the active AgentTurn, or by passing a CancellationToken from a CancellationController to chat.send() or chat.sendStream().

final controller = CancellationController();
final turn = chat.sendStream(
  text: 'Write a long report.',
  cancel: controller.token,
);

// Later, abort the turn:
controller.cancel('user cancelled'); // or: turn.abort();

final res = await turn.response;
print(res.finishReason.value); // 'aborted'

An aborted turn does not throw AgentError; it resolves with res.finishReason == AgentFinishReason.aborted. For a server-managed agent, the aborted turn commits a rerunnable snapshot (res.snapshotId) holding the last-good history before the turn.

When a turn fails after the invocation starts, the client throws AgentError. The exception carries details of the failure along with the last-good state (err.response.raw.state holds the full SessionState; err.state holds the typed custom state).

try {
  await chat.send(text: 'Use a broken tool.');
} on AgentError catch (err) {
  print(err.status);
  print(err.snapshotId);
  print(err.state);
}

For a server-managed agent, failed turns commit a rerunnable snapshot with the last-good state. Resume a failed snapshot with agent.chat(snapshotId: err.snapshotId) (or an aborted snapshot with agent.chat(snapshotId: res.snapshotId)) and send an empty message to re-drive the turn, or new text to steer the retry. See Agent error handling for the full recovery flow.

Initialization misuse, such as sending state to a server-managed agent, is rejected before a turn starts.

Local agents from ai.define_agent() and remote clients from remote_agent() share this interface, so the same code drives both.

chat = weather_agent.chat()
res = await chat.send('Weather in Tokyo?')

print(res.text)
print(res.session_id)
print(res.snapshot_id)
print(res.state)

Calling chat() without arguments starts a new conversation. Pass session_id to resume a server-managed conversation, snapshot_id when you need an exact saved point, or messages / state / artifacts when the client owns the full session state.

chat = weather_agent.chat(session_id='user-session-123')
await chat.send('What did we discuss last time?')

load_chat() reads a server snapshot and hydrates messages, custom state, artifacts, snapshot_id, and session_id before the next turn.

chat = await weather_agent.load_chat(session_id='user-session-123')

print(len(chat.messages))
print(chat.state)

await chat.send('Continue from there.')

Use get_snapshot() when you want read-only data (such as checking background task status, inspecting errors, or auditing session state) without opening a chat session. Use load_chat() when you want an interactive AgentChat instance to continue the conversation and send new turns.

chat = weather_agent.chat()
turn = chat.send_stream('Weather in Tokyo?')

async for chunk in turn.stream:
    if chunk.text:
        print(chunk.text, end='', flush=True)
    if chunk.custom is not None:
        update_status(chunk.custom)
    if chunk.artifact is not None:
        render_artifact(chunk.artifact)

res = await turn.response
print(res.finish_reason)

chat.send_stream() returns an AgentTurn. Iterate turn.stream for live chunks, or await turn.response for the final output. Both paths apply custom-state patches so chat.state stays current.

Cancel a foreground turn with turn.abort(). This stops the client from listening; for store-backed agents, call chat.abort() if you also need to stop server-side work.

turn = chat.send_stream('Write a long report.')

# Later:
await turn.abort()

res = await turn.response
print(res.finish_reason)  # AgentFinishReason.ABORTED

When a turn fails after the invocation starts, the client raises AgentError. The exception carries status, details, the latest snapshot ID, and the recoverable last-good state.

from genkit.agent import AgentError

try:
    await chat.send('Use a broken tool.')
except AgentError as err:
    print(err.status)
    print(err.snapshot_id)
    print(err.state)

Initialization misuse, such as sending state to a server-managed agent, raises AgentInitError before a turn starts.