Run and stream agents

Beta

The Agents API is in Beta. It can introduce breaking changes in minor version releases.

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 unified API

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(), waitForSnapshot(), 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.

Start a chat

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.

Pass context to an in-process agent

An agent from ai.defineAgent() accepts call options as the second argument to chat() and loadChat(). Use context to pass auth or other request data to the agent’s tools and middleware:

const chat = weatherAgent.chat({}, { context: { auth: { uid: 'alice' } } });
await chat.send('Weather in Tokyo?');

// A per-call `context` replaces the bound one entirely; the two are not merged.
await chat.send('And in Paris?', { context: { auth: { uid: 'bob' } } });

Options bound on chat() apply to every call on that chat, including detached tasks it starts, so auth carries through background work. remoteAgent() does not take context, because over HTTP the server derives it from the request; TypeScript rejects it at compile time.

Restore a full chat

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.

Stream a turn

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.

Abort a foreground turn

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. For an in-process agent, the chat keeps the turns that finished before the stop, so chat.send({}) continues from there. With remoteAgent, the abort cancels the HTTP request and the chat stays at its previous snapshot; reload the session with loadChat({ sessionId }) before you continue.

Failed turns

When a turn fails after the invocation starts, the client throws AgentError. The error carries the resume point and the response object when available: err.snapshotId for a server-managed agent, or the full state at err.response.raw.state for a client-managed one (err.state holds only the custom state). When the failed turn committed its completed tool rounds, the chat has already adopted that resume point, so chat.send({}) retries the turn. See Resume after a failure or a stop.

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.

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

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"`
    // Usage sums the agent's own model calls in this invocation, failed turns
    // included. Nil when the invocation detached.
    Usage *ai.GenerationUsage `json:"usage,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. See Token usage for Usage.

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"`
    // Usage sums the agent's own model calls in this turn, failed or not.
    Usage *ai.GenerationUsage `json:"usage,omitempty"`
}

Every turn ends with exactly one TurnEnd, a failed turn included, and it arrives before the output resolves. 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

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.

Constant	Wire value	Meaning
aix.AgentFinishReasonStop	stop	The model stopped naturally.
aix.AgentFinishReasonLength	length	Generation hit the token limit.
aix.AgentFinishReasonBlocked	blocked	Generation was blocked, usually by a safety filter. Only a custom agent that forwards the model’s reason reports it: a prompt-backed agent fails a blocked turn with ai.ErrGenerationBlocked. See Agent error handling.
aix.AgentFinishReasonInterrupted	interrupted	A tool paused for input. See Agent interrupts.
aix.AgentFinishReasonOther	other	The model stopped for some other reason.
aix.AgentFinishReasonUnknown	unknown	The model gave no reason.
aix.AgentFinishReasonAborted	aborted	The caller stopped the run: a cancelled context, an expired deadline, a closed transport, a limit it set such as ai.WithMaxTurns or a usage budget, or Abort on a detached run. The snapshot keeps the turns that finished.
aix.AgentFinishReasonDetached	detached	The client detached and the work continues in the background.
aix.AgentFinishReasonFailed	failed	A turn broke. Read out.Error. The snapshot keeps the tool rounds the turn completed.

Token usage

An agent reports what its own model calls used at three levels. Each is an *ai.GenerationUsage, summed field by field:

• chunk.TurnEnd.Usage covers one turn, whether it succeeded or failed.
• out.Usage covers one invocation, failed turns included. It is nil when the invocation detached, because the work continues.
• SessionState.Usage covers the whole session. Snapshots persist it, so read it from snap.State.Usage for a server-managed agent or from out.State.Usage for a client-managed one.

for chunk, err := range conn.Receive() {
    if err != nil {
        return fmt.Errorf("stream turn: %w", err)
    }
    if chunk.TurnEnd != nil && chunk.TurnEnd.Usage != nil {
        fmt.Printf("turn used %d input and %d output tokens\n",
            chunk.TurnEnd.Usage.InputTokens, chunk.TurnEnd.Usage.OutputTokens)
        break
    }
}

out, err := conn.Output()
if err != nil {
    return fmt.Errorf("finalize turn: %w", err)
}
if out.Usage != nil {
    fmt.Printf("invocation used %d input tokens\n", out.Usage.InputTokens)
}

The session usage is a ledger: each model call adds to it when the call returns, and nothing takes it back. A failed turn’s messages can roll back, but its spend stays on the session. Usage counts the agent’s own calls only. A sub-agent records its calls in its own session, so a sum of out.Usage across agents never counts a token twice. A tool that calls genkit.Generate directly, with no agent in between, counts toward the turn that ran it.

To stop a run or a session that spends too much, attach the Budget middleware. See Middleware.

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

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 also ends the run as aborted, because a limit the caller set is a caller stop rather than a failure, but it reports in band: Run returns the output with a nil error, and out.Error carries ai.ErrMaxTurnsExceeded. A custom agent should still check ctx.Err() between turns.

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

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

• 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

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

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

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(), Description(), Ref(), 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.

A handle does not care where the agent runs. genkitx.DefineRemoteAgent returns a handle to an agent that another Genkit app serves over HTTP, with the same calls, and registers it so LookupAgent and the agents middleware find it by name. See Call a remote agent.

Next steps

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

Start a chat

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?');

Restore a full chat

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.

Stream a turn

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.

Pass per-turn context

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.

Abort a foreground turn

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.

Failed turns

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() (in genkit.exp) and remote clients from remote_agent() (in genkit.exp.agent) share this interface, so the same code drives both. weather_agent below is the agent ai.define_agent() returned; see Define agents.

Start a chat

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

print(res.text)
# => It is 72°F and sunny in Tokyo.
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?')

Restore a full chat

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

load_chat() points the chat at the stored row as is. If the latest turn failed or was aborted, the next send() on that chat returns a failed response with INVALID_ARGUMENT. Use chat(session_id=...) to continue from the last completed turn instead.

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.

Stream a turn

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)
# => stop

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.

Abort a foreground turn

Call await turn.abort() to stop listening to a foreground turn. The client detaches, and the server turn keeps running in the background.

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

async for chunk in turn.stream:
    if chunk.text:
        print(chunk.text, end='', flush=True)
    if user_pressed_stop():
        await turn.abort()
        break

After turn.abort(), await turn.response raises asyncio.CancelledError. Cancelling the task or hitting an asyncio.timeout() deadline (Python 3.11+) detaches the client the same way and raises CancelledError or TimeoutError.

To stop server-side work on a store-backed agent, run the turn with chat.detach() and call await task.abort(). See Background execution.

Failed turns

When a turn fails during execution, such as a model error or an unhandled tool exception, chat.send() and turn.response do not raise. They return an AgentResponse with finish_reason == AgentFinishReason.FAILED and res.error set to a Failure (status, message, details). The chat drops the failed turn’s user message and custom-state patches, and the next chat.send() continues from the last completed snapshot (or client state).

res = await chat.send('Use a broken tool.')

if res.error:
    print(res.error.status, res.error.message)
    # => INTERNAL internal error

    # The chat rolled back, so this turn continues from the last completed one.
    res = await chat.send('Try a fallback approach.')

chat.send() raises GenkitError only when no turn result arrives. Init misuse that the client can detect, such as passing state to a store-backed agent, raises AgentInitError (a GenkitError subclass) from agent.chat() itself, before any turn starts. See Agent error handling for recovery patterns.