Sessions and state

In Genkit, agent state includes message history, custom application state, artifacts, session identity, and snapshot lineage. Choose the state strategy before building the client because it determines who owns continuity between turns.

State strategies

Genkit agents can keep conversation continuity in one of two ways:

Server-managed state means the agent has a store (for example InMemorySessionStore, FileSessionStore, or FirestoreSessionStore). The server persists messages, custom state, and artifacts between turns. Clients continue the chat by sending a session_id (loads the latest completed state) or snapshot_id (loads exact state). Use this for long-running chat applications, multi-turn generators, background detached tasks, or any workflow where client payload size should remain small.

Client-managed state means the agent has no store. The server returns the full session at the end of each turn. To resume in a new chat, pass it back with chat(messages=..., state=..., artifacts=...). Use this when your client app or another system already owns persistence, when you need stateless server deployments, or when you want to encrypt conversation data outside Genkit.

In both modes, AgentChat tracks continuity for you. Server-managed chats carry forward a session_id / snapshot_id. Client-managed chats carry forward messages, custom state, and artifacts.

Understand session state

Session state has three main fields:

custom is your application-specific data. Use it for lightweight status variables, preferences, selected options, or list state that tools or the model needs across turns. Keep this payload small as it is serialized in every snapshot and client payload.
messages is the accumulated message history. The runtime appends user prompts, tool calls, and model replies automatically.
artifacts is a list of structured, versioned outputs generated by the agent or tools. Use artifacts for larger outputs that the user might download or inspect independently.

Custom state vs. artifacts

Use custom state for lightweight control parameters and variables that direct the agent’s next model call or UI rendering. Keep it small.
Use artifacts for files, patches, or comprehensive reports that the agent produces. Do not store large documents directly in custom state.

Modify custom state

Tools and custom agents read the active session with current_session() from genkit.exp and update custom state with await sess.update_custom(fn). The updater receives the current custom state, which is a plain dict when it was loaded from a snapshot or client payload, so validate it before use. With a state_schema, chat.state, res.state, and streamed chunk.custom come back as that Pydantic model. Genkit diffs the result and streams RFC 6902 JSON Patches to the client mid-turn.

from typing import Any

from pydantic import BaseModel

from genkit.exp import Genkit, current_session
from genkit_google_genai import GoogleAI

ai = Genkit(plugins=[GoogleAI()])

class TaskItem(BaseModel):
    id: int
    title: str
    done: bool = False

class TaskState(BaseModel):
    tasks: list[TaskItem] = []
    next_id: int = 1

@ai.tool()
async def add_task(title: str) -> str:
    """Add a task to the user's list."""
    sess = current_session()
    if sess is None:
        raise RuntimeError('add_task requires an active agent session.')

    created = TaskItem(id=0, title=title)

    def mutate(custom: dict[str, Any] | None) -> dict[str, Any]:
        state = TaskState.model_validate(custom or {})
        created.id = state.next_id
        state.tasks.append(created)
        state.next_id += 1
        return state.model_dump()

    await sess.update_custom(mutate)
    return f'Added task #{created.id}: {title}'

task_agent = ai.define_agent(
    name='taskAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='Manage the user task list with add_task.',
    tools=[add_task],
    state_schema=TaskState,
)

Server-managed stores

Add a store when defining your agent to enable server-managed persistence:

from genkit.exp.agent import FileSessionStore

store = FileSessionStore('.sessions')

weather_agent = ai.define_agent(
    name='weatherAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='You are a helpful weather assistant.',
    store=store,
)

On every successful turn, the store saves a completed snapshot capturing the conversation’s exact state. A failed turn saves a failed snapshot with the error on snapshot.error, and the chat keeps resuming from the last completed snapshot. See Run and stream for how a failed turn reaches the caller.

For store choices, see Session stores.

Snapshots

Read a snapshot directly by ID or fetch the latest snapshot in a session:

snapshot = await weather_agent.get_snapshot(snapshot_id='snapshot-123')
latest = await weather_agent.get_snapshot(session_id='session-123')

A session_id lookup returns the newest row, even when that turn failed or was aborted.

Snapshot lifecycle

pending: A detached background turn is still running.
completed: The snapshot captures a settled, resumable state.
failed: The turn failed. snapshot.error holds the Failure details.
aborted: The detached turn was canceled.
expired: A pending snapshot’s heartbeat went stale, so the background worker is presumed dead. Stores report this at read time.

InMemorySessionStore, FileSessionStore, and FirestoreSessionStore move an aborted row straight from pending to aborted. aborting appears only on rows another SDK wrote to a shared store.

Only completed snapshots are valid resume points. Other statuses are useful for UI progress indicators and background tracking.

Resume by session or snapshot

To continue a server-managed conversation, pass the session_id. If the newest row is failed, aborted, or pending, the agent walks back to the last completed snapshot:

chat = weather_agent.chat(session_id='user-session-123')
await chat.send('What is the weather in Tokyo?')

To branch from a specific historical point, pass the snapshot_id to chat(), or use load_chat(snapshot_id=...) to hydrate messages, state, and artifacts before the next turn:

branch = await weather_agent.load_chat(snapshot_id='snapshot-abc-456')
await branch.send('Check Osaka instead.')

load_chat() loads the row as stored. A send() from a row that is not completed returns a failed turn.

Branching creates a second leaf in the session. InMemorySessionStore and FileSessionStore resolve a session_id to the most recent leaf. Pass reject_ambiguous_session=True to either store to reject session_id lookups on a forked session and require a snapshot_id.

Client-managed state

If your agent does not use a server store, capture messages, custom state, and artifacts yourself, then pass them back:

chat = task_agent.chat()
await chat.send('Add buy milk to my tasks.')

messages, state, artifacts = chat.messages, chat.state, chat.artifacts

resumed = task_agent.chat(messages=messages, state=state, artifacts=artifacts)
res = await resumed.send('What is on my list?')
print(res.text)
# => Your list has one task: buy milk.

Live custom state

When custom state changes during a turn, the runtime streams incremental RFC 6902 JSON Patch chunks. AgentChat applies them automatically, yielding the updated state on chunk.custom and chat.state.

turn = task_agent.chat().send_stream('Add buy milk and walk the dog.')

async for chunk in turn.stream:
    if chunk.custom is not None:
        print([task.title for task in chunk.custom.tasks])
        # => ['buy milk']
        # => ['buy milk', 'walk the dog']

res = await turn.response

The first patch emitted in a turn is a whole-document replace that aligns the client’s state baseline with the server’s.

Artifacts

Record artifacts from tools or custom agents with sess.add_artifacts():

from genkit import Part
from genkit.exp import current_session
from genkit.exp.agent import Artifact

@ai.tool()
async def save_report(markdown: str) -> str:
    """Save a markdown research report as a session artifact."""
    sess = current_session()
    if sess is None:
        raise RuntimeError('save_report requires an active agent session.')
    await sess.add_artifacts([Artifact(name='report.md', parts=[Part.from_text(markdown)])])
    return 'Saved report.md'

Artifacts with identical names overwrite earlier ones, while unnamed artifacts are appended to the session.

To let the model read and write artifacts itself, add the experimental Artifacts middleware from genkit_middleware.exp to the agent’s use list. It adds read_artifact and write_artifact tools and lists current artifacts in the system prompt. Pass Artifacts(readonly=True) to expose only read_artifact.

Client transforms

Use state_transform and chunk_transform to redact or reshape what leaves the server before it reaches a client. state_transform shapes snapshot reads, client-managed output, and the baseline for streamed custom patches, and must always return a state. Returning None from chunk_transform drops the chunk:

from genkit.exp.agent import AgentStreamChunk, SessionState

def redact_state(state: SessionState) -> SessionState:
    # Drop the internal account token before the client sees it.
    if isinstance(state.custom, dict):
        custom = {k: v for k, v in state.custom.items() if k != 'internal_token'}
        return state.model_copy(update={'custom': custom})
    return state

def hide_internal_notes(chunk: AgentStreamChunk) -> AgentStreamChunk | None:
    if chunk.artifact is not None and chunk.artifact.name == 'internal-notes.md':
        return None
    return chunk

support_agent = ai.define_agent(
    name='supportAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='Help customers with account questions.',
    state_transform=redact_state,
    chunk_transform=hide_internal_notes,
)