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.

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

Start a chat

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

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

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)

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

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

Failed turns

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.