claude-agent-sdk-mongodb-session-store

MongoDB SessionStore for the Claude Agent SDK backed by the official pymongo driver via its stable async API pymongo.AsyncMongoClient.

Installation

uv add claude-agent-sdk-mongodb-session-store

This requires pymongo 4.14 or later, which is installed as a dependency.

Usage

from pymongo import AsyncMongoClient
from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query

from claude_agent_sdk_mongodb_session_store import MongoDBSessionStore

uri = "mongodb://localhost:27017"  # local development. Use TLS and auth in production
client = AsyncMongoClient(
    uri,
    readPreference="primary",
    w="majority",
    serverSelectionTimeoutMS=10_000,
    timeoutMS=20_000,
)
store = MongoDBSessionStore(client=client, db_name="claude")
await store.create_schema()  # idempotent createIndexes

async for message in query(
    prompt="Hello!",
    options=ClaudeAgentOptions(session_store=store),
):
    if isinstance(message, ResultMessage) and message.subtype == "success":
        print(message.result)

The constructor also takes entries_collection, summaries_collection and counters_collection (defaults claude_session_entries, claude_session_summaries, claude_session_counters), or a single options=MongoDBSessionStoreOptions(...). A collection name MongoDB rejects, or one in the reserved system. namespace, raises ValueError.

Every method checks its key before touching the database: project_key, session_id and subpath must be strings (TypeError otherwise), and project_key and session_id must be non-empty (ValueError). See Security for why.

Client configuration

The store sets no timeouts, read preference or write concern of its own; it uses whatever the AsyncMongoClient you pass in is configured with. Its availability is that of your MongoDB deployment. We recommend readPreference=primary, w=majority, serverSelectionTimeoutMS of about 10 s and timeoutMS of about 20 s as shown in Usage above.

Use the primary for reads, because load() for a resume and the summary rebuild in append() must see the latest entries. w="majority" keeps an acknowledged append from being rolled back by a failover. Keep the timeouts below the SDK's 60 s limit on each append(). The SDK retries a failed append() up to three times, but not one that hits its own 60 s limit, because the call may still land. A driver timeout (NetworkTimeout, ExecutionTimeout) is retried, and the entry uuid keeps the retry from storing a duplicate (see Retries). timeoutMS applies to each driver operation, and one append() makes several, so it bounds a call but does not cap it at 20 s.

Writes need the primary, so appends fail during an election. The SDK drops a batch it cannot append and the agent run continues, but load() fails after the SDK's load_timeout_ms (60 s by default). Inserts, the counter update and the summary write are retryable writes. delete() and delete_inactive() use delete_many, which is not, so a failover can stop them partway; run them again to finish.

Retries

The SDK retries a failed append() with the same batch; see Client configuration for which failures it retries. Most entries carry a uuid, which the adapter treats as an idempotency key: an entry whose uuid is already stored in the same transcript is skipped, so a retry never stores it twice. The same holds when import_session_to_store() re-imports a session. Entries without a uuid (titles, tags, mode markers) are stored every time, as the SDK's SessionStore protocol asks.

Concurrency

Per the SessionStore.list_session_summaries contract, sidecar updates inside append() must be serialized when calls can race for the same session. (The sidecar is the per-session summary document; see Schema and Why a summary sidecar?.) The summary write is a compare-and-swap on its rev token, which changes on every write, so appends from different processes (workers, pods) to the same session can't overwrite each other's summary. A writer that loses the race reads the summary again and rebuilds it from the stored entries. The same check stops an in-flight append from restoring the summary of a session that was deleted in the meantime. If the summary keeps changing, the write gives up after 16 attempts and append() raises RuntimeError.

The store keeps no per-session state in memory, so one instance can serve any number of sessions in a long-running process.

Retention

This adapter never deletes documents on its own. Run delete_inactive() on a schedule (cron, an Atlas scheduled trigger, ...) to delete every session with no appends for a given period:

from datetime import timedelta

deleted = await store.delete_inactive(timedelta(days=30))

A session is active if any of its transcripts, main or subagent, was appended to within the period. Sessions are deleted whole, in every project. The sweep reads only the counters collection, one small document per transcript, never the entries.

Don't expire individual entries with a TTL index or a delete_many({"mtime": {"$lt": cutoff}}). Both delete the oldest entries of a session that is still in use, leaving a transcript that can't be resumed. A TTL index would also do nothing here: TTL needs a BSON Date, and mtime is an integer.

delete() is implemented (cascades to subpath documents and the summary sidecar) but is only called when you invoke delete_session_via_store() from the SDK.

Local-disk transcripts under CLAUDE_CONFIG_DIR are swept independently by the CLI's cleanupPeriodDays setting.

Resume from MongoDB

async for message in query(
    prompt="Continue where we left off",
    options=ClaudeAgentOptions(
        session_store=store,
        resume="previous-session-id",
    ),
):
    ...

Resuming replays the stored transcript as the agent's history, so only resume from a database that only your application can write to. See Security.

Security

Transcripts are sensitive. Every entry is stored verbatim: user prompts, assistant replies, tool inputs and tool results. Tool results routinely include file contents, command output and environment variables, so a transcript can contain anything the agent saw, including credentials. The summary sidecar's data blob (a per-session summary document; see Schema) holds a digest of the same material. Treat the three collections as you would any store of user conversations:

• Connect over TLS and authenticate. Both are configured on the AsyncMongoClient URI the caller passes in; the store never reads credentials itself. On Atlas, both are on by default.
• Give the application a dedicated database user with the readWrite role on this one database and nothing else. create_schema() needs createIndex, which readWrite includes.
• Write access to these collections is equivalent to controlling what a resumed session believes.
• Don't resume sessions from a database that untrusted parties can write to.
• Turn on encryption at rest. If plaintext transcripts must never reach the server, configure the AsyncMongoClient for automatic Client-Side Field Level Encryption (CSFLE) with randomized encryption of the entire entry and data objects. The store never queries inside these payloads, so they do not need to be queryable. Keep identifiers and indexed metadata fields unencrypted.
• The store doesn't authenticate callers or sign entries. The application must derive project_key and session_id from an authenticated identity and never take them from untrusted input.

Keys are validated before every query. project_key, session_id and subpath are passed into MongoDB query filters. A value that is not a string, such as the dict {"$ne": ""}, would otherwise act as a query operator there and match other sessions, or other tenants when project_key is a tenant id. The store raises TypeError for a non-string and ValueError for an empty project_key or session_id before any I/O. The SDK already validates session_id as a UUID on its own code paths; this check protects callers that drive the store directly, for example from a web handler.

delete_inactive() spans every tenant. It sweeps all project_keys in the database. Run it from a privileged maintenance job, not from code that acts on behalf of one tenant.

Reads are unbounded. load() returns a whole transcript, and append() rebuilds the summary from the whole main transcript whenever it loses the compare-and-swap or sees a duplicate, up to 16 times per call. Very large sessions under heavy write contention can make a single append expensive. Use delete_inactive() to keep the data set bounded.

Production checklist

• Size the AsyncMongoClient connection pool for expected concurrent sessions; don't share a pool with request-handler code that holds connections.
• Schedule delete_inactive() for retention — every collection grows unbounded otherwise.
• Work through the Security section: TLS and auth on the URI, a least-privilege database user, and encryption at rest.
• Set the client configuration: primary reads, w="majority" and timeouts below the SDK's 60 s limit.

Schema

A transcript is the record of one conversation that Claude Code writes as it runs: a JSONL file with one entry per line (user prompts, assistant replies, tool calls and results, plus metadata such as a custom title). A session has a main transcript and one more for each subagent it starts. Locally they live at ~/.claude/projects/<project_key>/<session_id>.jsonl and <session_id>/subagents/agent-<id>.jsonl.

In the store, a transcript is identified by (project_key, session_id, subpath), where subpath is "" for the main transcript and, for example, subagents/agent-1 for a subagent's. Three collections share a single database.

Entries — one document per transcript entry, ordered by position:

{
  "_id": ObjectId,
  "project_key": str,
  "session_id":  str,
  "subpath":     str,    # "" for the main transcript
  "position":    int,    # per-transcript append order
  "uuid":        str,    # entry["uuid"], when it has one
  "entry":       <opaque JSON>, # single jsonl line
  "mtime":       int,    # Unix epoch ms, write-time stamp
}

Summaries — one document per main session, maintained incrementally inside append() via fold_session_summary:

{
  "_id":           {"project_key": str, "session_id": str},
  "mtime":         int,       # epoch ms (same clock as entries)
  "last_position": int,       # last main-transcript entry folded in
  "rev":           ObjectId,  # new on every write, for compare-and-swap
  "data":          <opaque>,  # SDK-owned summary state, persisted verbatim
}

Counters — one document per transcript. It guarantees the order of entries written by different processes: each append() reserves the next block of positions from it. It also records the time of the latest append, which delete_inactive() reads instead of scanning entries. delete() keeps it, so an append racing the delete can't restart positions at 1; delete_inactive() removes it once idle:

{
    "_id": {"project_key": str, "session_id": str, "subpath": str},
    "last_position": int,
    "mtime": int,  # epoch ms (same clock as entries)
}

MongoDB's default _id is an ObjectId generated by the client (pymongo) from the client's clock, so it cannot guarantee order of entries written by different processes. Instead, append() reserves a block of positions with one atomic $inc on the counter document, and load() is find().sort("position", 1).

create_schema() creates five indexes — a unique (project_key, session_id, subpath, position) serving load()/delete()/list_subkeys(), a unique (project_key, session_id, subpath, uuid) over entries that have a uuid, (project_key, subpath, session_id, mtime DESC) that lets list_sessions() read one index key per session (a DISTINCT_SCAN), (_id.project_key) on the summaries collection, and (_id.project_key, _id.session_id) on the counters collection for delete_inactive().

Why a summary sidecar? The optional list_session_summaries method

Unlike the S3, Redis, and Postgres reference adapters in the SDK's examples/session_stores/ directory, this adapter implements the optional list_session_summaries method. list_sessions_from_store() takes a fast path when a store offers it — one batch read for all summaries plus a cheap list_sessions() to gap-fill stale or missing entries — instead of falling back to N per-session load() calls (bounded at 16 concurrent) to recompute the summary on the fly. For projects with many sessions or remote-backend latency, that's the difference between one round trip and dozens.

The summary itself is computed by the SDK's fold_session_summary (read inside append() and written back as one opaque data blob); the adapter never interprets the contents. fold_session_summary is exported from claude_agent_sdk as of 0.1.65.

Normally append() folds only its own batch into the summary. The summary records the last entry position it has folded in. If a batch doesn't directly follow that position, or some of its entries were skipped as duplicates (a retry, a re-sent entry, or appends finishing out of order), the summary is rebuilt from the stored transcript instead. The summary's mtime never moves backwards and matches the newest entry's.

Live MongoDB end-to-end

There is no in-process MongoDB mock that faithfully exercises aggregation, distinct, or the query planner, so every test that touches the store runs live-only. Those tests skip unless MONGODB_URI is set; the constructor and validation tests always run.

docker run -d -p 27017:27017 mongo:latest

MONGODB_URI=mongodb://localhost:27017 uv run pytest -v

Each test uses a random database name and drops it on teardown. The query-plan tests turn on the database profiler for their own database only.

Credits

The adapter was originally written by Alex Bevilacqua as a reference example in anthropics/claude-agent-sdk-python#1014; this repository's early history replays those commits with their original authorship. That example was itself a port of the MongoDBSessionStore reference adapter in the TypeScript SDK's examples/session-stores/mongodb/.

License

Apache-2.0. See LICENSE.