Metadata-Version: 2.5
Name: claude-agent-sdk-mongodb-session-store
Version: 0.1.0
Summary: MongoDB-backed SessionStore for the Claude Agent SDK
Project-URL: Homepage, https://github.com/mongodb-developer/claude-agent-sdk-mongodb-session-store
Project-URL: Changelog, https://github.com/mongodb-developer/claude-agent-sdk-mongodb-session-store/blob/main/CHANGELOG.md
Project-URL: Issues, https://jira.mongodb.org/projects/INTPYTHON
Author-email: Casey Clements <casey.clements@mongodb.com>
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: ai,anthropic,claude,mongodb,sdk,session-store
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Database
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: claude-agent-sdk>=0.1.66
Requires-Dist: pymongo>=4.14
Description-Content-Type: text/markdown

# claude-agent-sdk-mongodb-session-store

MongoDB `SessionStore` for the [Claude Agent SDK](https://github.com/anthropics/claude-agent-sdk-python) backed by the official [`pymongo`](https://www.mongodb.com/docs/drivers/pymongo/)
driver via its stable async API `pymongo.AsyncMongoClient`.

## Installation

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

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

## Usage

```python
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](#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](#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](#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](#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](#schema) and [Why a summary sidecar?](#why-a-summary-sidecar-the-optional-list_session_summaries-method).)
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:

```python
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

```python
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).

## 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](#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](https://www.mongodb.com/docs/manual/core/csfle/fundamentals/encryption-algorithms/)
  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_key`s 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](#security) section: TLS and auth on the URI, a
  least-privilege database user, and encryption at rest.
- Set the [client configuration](#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`:

```python
{
  "_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`:

```python
{
  "_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
`position`s 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:

```python
{
    "_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.

```bash
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](https://github.com/anthropics/claude-agent-sdk-python/pull/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/`](https://github.com/anthropics/claude-agent-sdk-typescript/tree/main/examples/session-stores/mongodb).

## License

Apache-2.0. See [LICENSE](LICENSE).
