Session stores

Session stores persist snapshots for server-managed agents. They are the storage layer behind sessionId and snapshotId resumption, snapshot reads, branching, background execution, and aborting detached work.

Use the Sessions and state guide first when you are deciding between server-managed and client-managed state. Use this page when you know the server should own state and need to choose or implement the persistence layer.

Choose a store

In-memory store (InMemorySessionStore from genkit.exp.agent) for tests, local examples, command-line interfaces, and single-process experiments.
File store (FileSessionStore from genkit.exp.agent) for local development, prototypes, and single-host applications where snapshots must persist across process restarts.
Firestore store (FirestoreSessionStore from genkit_google_cloud.exp) for production apps where several server instances share sessions.
Custom store (implementing SessionStore from genkit.exp.agent) when you need a different shared database such as Cloud SQL, Spanner, Postgres, or Redis.

Configure the store on the agent. The runtime handles reads and writes. With HTTP serving, /getSnapshot and /abort are mounted only when a store is configured.

Use an in-memory store

InMemorySessionStore keeps snapshots in local process memory. It is fast and requires no setup.

from genkit.exp.agent import InMemorySessionStore

store = InMemorySessionStore()

support_agent = ai.define_agent(
    name='supportAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='Help customers with their orders.',
    store=store,
)

Do not use the in-memory store if session history must survive process restarts or when scaling horizontally across multiple server instances.

Both built-in local stores accept reject_ambiguous_session=True. Branching from an earlier snapshot gives a session a second leaf; with this flag set, a session_id lookup on a forked session raises instead of picking the newest leaf, so callers must resume by snapshot_id.

Use a file-backed store

FileSessionStore writes one <snapshot_id>.json file per snapshot inside a local directory. This is the standard choice for local development or single-host deployments.

from genkit.exp.agent import FileSessionStore

store = FileSessionStore('.sessions', max_persisted_chain_length=50)

weather_agent = ai.define_agent(
    name='weatherAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='You help with weather questions.',
    store=store,
)

max_persisted_chain_length caps how many snapshots of one conversation stay on disk. Once a chain grows past it, each save deletes the oldest turns. Resuming still works, but you cannot rewind or branch past the retained window. Leave it unset to keep the full history.

Use a Firestore store

FirestoreSessionStore from genkit_google_cloud.exp (genkit-google-cloud package) persists snapshots in Cloud Firestore. It is the built-in option for multi-instance production deployments, such as Cloud Run, that share sessions across instances.

Storage model

Diffs and sharded checkpoints: Saves each turn as an RFC 6902 JSON Patch diff from its parent and writes a full-state checkpoint every checkpoint_interval turns. Checkpoints split across shard documents, so long sessions stay under Firestore’s 1 MiB document limit.
Direct document-ID reads: Resolves sessions through a pointer document and direct document-ID fetches inside a read transaction. No composite indexes are required.
Transactional writes: Commits snapshot documents and session pointers atomically, retrying contended pointer updates up to transaction_max_attempts times before raising GenkitError with status ABORTED.
Cross-instance status watches: Streams status changes through Firestore document watches, so task.wait(), task.poll(), and task.abort() work across instances.

from typing import Any

from genkit_google_cloud.exp import FirestoreSessionStore

def by_user(context: dict[str, Any] | None) -> str:
    uid = (context or {}).get('uid')
    return uid if isinstance(uid, str) else 'global'

store = FirestoreSessionStore(
    collection='genkit-sessions',
    snapshot_path_prefix=by_user,
)

support_agent = ai.define_agent(
    name='supportAgent',
    model=GoogleAI.gemini('gemini-flash-latest'),
    system='Help customers check order status.',
    store=store,
)

Constructor options

All options are keyword-only and optional:

client: Explicit google.cloud.firestore.AsyncClient. Defaults to a client using Application Default Credentials, or FIRESTORE_EMULATOR_HOST when set.
sync_client: Explicit synchronous google.cloud.firestore.Client for document watches. When omitted, the store creates one per event loop and closes it in await store.close().
collection: Base collection name (defaults to 'genkit-sessions'). The store derives <collection>-pointers and <collection>-shards. Document formats differ by SDK, so give each SDK its own collection.
snapshot_path_prefix: Function that maps the call context to a tenant path segment, such as the authenticated user ID (defaults to 'global').
checkpoint_interval: Turns between full-state checkpoints (defaults to 25).
shard_size: Maximum bytes per shard document (defaults to 512 KiB).
transaction_max_attempts: Retry budget for contended pointer transactions (defaults to 5).

The Firestore store does not prune snapshots, so plan retention for long-running or artifact-heavy conversations as described in Production guidance.

Implement a production store

When the built-in stores do not fit, implement the SessionStore protocol from genkit.exp.agent. Add SnapshotSubscriber when you need detach and abort. The exported helpers keep the runtime’s invariants:

import asyncio
from typing import Any

from genkit.exp.agent import (
    TERMINAL_STATUSES,
    SaveFn,
    SessionSnapshot,
    SessionStore,
    SnapshotStatus,
    SnapshotStatusStream,
    SnapshotSubscriber,
    apply_save,
    iterate_statuses,
    require_one_selector,
)

class DatabaseSessionStore(SessionStore[Any], SnapshotSubscriber):
    def __init__(self) -> None:
        self._lock = asyncio.Lock()
        self._subs: dict[str, list[asyncio.Queue[SnapshotStatus | None]]] = {}

    async def get_snapshot(
        self,
        *,
        snapshot_id: str | None = None,
        session_id: str | None = None,
        context: dict[str, Any] | None = None,
    ) -> SessionSnapshot | None:
        require_one_selector(snapshot_id=snapshot_id, session_id=session_id)
        # 1. Load by snapshot_id, or resolve the latest leaf for session_id.
        ...

    async def save_snapshot(
        self,
        snapshot_id: str,
        fn: SaveFn,
        *,
        context: dict[str, Any] | None = None,
    ) -> SessionSnapshot | None:
        # 2. Read, mutate, and write in one transaction (or under a lock).
        async with self._lock:
            existing = await self._read(snapshot_id)
            updated = apply_save(existing=existing, snapshot_id=snapshot_id, fn=fn)
            if updated is not None:
                await self._write(snapshot_id, updated)
                self._notify(snapshot_id, updated.status)
            return updated

    async def on_snapshot_status_change(
        self,
        snapshot_id: str,
        *,
        context: dict[str, Any] | None = None,
    ) -> SnapshotStatusStream:
        # 3. Seed the current status, then stream changes until the snapshot settles.
        queue: asyncio.Queue[SnapshotStatus | None] = asyncio.Queue()
        async with self._lock:
            current = await self._read(snapshot_id)
            if current is not None:
                _push(queue, current.status)
            self._subs.setdefault(snapshot_id, []).append(queue)

        async def unsubscribe() -> None:
            self._subs[snapshot_id].remove(queue)

        return iterate_statuses(queue, on_close=unsubscribe)

    def _notify(self, snapshot_id: str, status: SnapshotStatus | None) -> None:
        for queue in self._subs.get(snapshot_id, []):
            _push(queue, status)

    async def _read(self, snapshot_id: str) -> SessionSnapshot | None: ...

    async def _write(self, snapshot_id: str, snapshot: SessionSnapshot) -> None: ...

def _push(queue: asyncio.Queue[SnapshotStatus | None], status: SnapshotStatus | None) -> None:
    queue.put_nowait(status)
    if status in TERMINAL_STATUSES:
        queue.put_nowait(None)  # End the stream.

get_snapshot fetches a snapshot by exact snapshot_id or resolves the latest snapshot for session_id. require_one_selector rejects calls that pass both or neither.
save_snapshot must be atomic. Run the read, mutator, and write inside one database transaction, or synchronize with a lock. The mutator (SaveFn) is synchronous and side-effect free, since stores may call it more than once under contention. apply_save runs it, stamps the store fields, and rejects status changes once a snapshot reaches a terminal status (completed, failed, or aborted).
on_snapshot_status_change is an async def that returns the stream. The runtime subscribes when a turn detaches and cancels the background work when the snapshot flips to aborted. Seed the current status so an abort written before the subscription is not missed, and put None after a terminal status so iterate_statuses ends the stream. Without SnapshotSubscriber, chat.detach() and task.wait() still work, but task.abort() only marks the snapshot aborted while the background turn runs to completion.
Multiple instances: The in-process queues above only see writes made through this store object. To abort a turn running on another instance, feed _notify from your database’s change stream, such as Postgres LISTEN/NOTIFY.

Production guidance

Security: Treat snapshots as sensitive user data. They can contain raw message history, tool results, and personal information. Scope reads, writes, and status subscriptions by the caller identity in context (for example with snapshot_path_prefix on FirestoreSessionStore) so a leaked snapshot_id cannot read or abort another user’s session.
Payload size: Because snapshots contain full conversational checkpoints, their size grows over long sessions. Plan database indexes and cleanup or archival routines before launching production systems.