Integrate MongoDB with the Claude Agent SDK

Overview

The claude-agent-sdk-mongodb-session-store package provides a MongoDB-backed session store adapter for Claude Agent SDK conversations in your application. A session is a record of the conversation and the actions an agent takes, stored as individual entries. By default, the Claude Agent SDK stores the transcript on the host that invokes the agent. The MongoDB adapter also saves those entries to MongoDB, so the transcript is kept both on the host and in MongoDB. You can then resume the session from another host.

You can use the adapter to perform the following tasks:

• Resume a session from a different process or host

• Keep session history in the same deployment as your application data

• Query and audit session history with MongoDB tools

Installation and Setup

To use the adapter, you must have the following components available:

• Python 3.10 or later

• A MongoDB deployment that runs MongoDB 7.0 or later

• An Anthropic API key

To learn how to get and set your Anthropic API key, see the Claude Agent SDK Quickstart. The session store does not use or see this key.

In your terminal, run the following command to install the adapter and its dependencies:

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

This command installs the adapter and its claude-agent-sdk and pymongo dependencies.

Set your connection string in an environment variable, as shown in the following example. The adapter uses the client that you create, so your credentials stay in your client configuration:

export MONGODB_URI="<connection-string>"

Persist a Session

By default, the Claude Agent SDK keeps each session transcript only on the host where the agent runs. The following example creates a MongoDB session store, initializes its indexes, and runs a query. The adapter persists the session transcript to MongoDB as the agent runs. The example prints the result and the session ID that the SDK assigns.

import asyncio
import os
from pymongo import AsyncMongoClient
from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from claude_agent_sdk_mongodb_session_store import MongoDBSessionStore
async def main() -> None:
    client = AsyncMongoClient(
        os.environ["MONGODB_URI"],
        readPreference="primary",
        w="majority",
        serverSelectionTimeoutMS=10_000,
        timeoutMS=20_000,
    )
    store = MongoDBSessionStore(client=client, db_name="claude")
    # Set up the database indexes that optimize the store's performance.
    # This operation is idempotent, so you can run it on every startup.
    await store.create_schema()
    async for message in query(
        prompt="Review this codebase and explain what it does.",
        options=ClaudeAgentOptions(session_store=store),
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)
            # The SDK assigns the session ID. Save it so that you can
            # resume the session later.
            print(f"Session ID: {message.session_id}")
    await client.close()
asyncio.run(main())

Resume a Session

To resume a session, create a ClaudeAgentOptions object and pass the store object and session ID of the session that you want to resume. The SDK returns the session ID in the session_id field of the first run's result message.

You can use the following snippet inside an async function to resume a session. Replace <session-id> with your session ID:

# Pass the same store and the session ID from the earlier run.
async for message in query(
    prompt="Suggest how to improve this codebase.",
    options=ClaudeAgentOptions(
        session_store=store,
        resume="<session-id>",
    ),
):
    if isinstance(message, ResultMessage) and message.subtype == "success":
        print(message.result)

Retention and Operational Considerations

The adapter stores each transcript entry as a document and never deletes sessions on its own. To remove sessions that are no longer active, call delete_inactive() on a schedule, such as from a cron job or an Atlas scheduled trigger:

from datetime import timedelta
# Delete every session with no new entries in the last 30 days.
deleted = await store.delete_inactive(timedelta(days=30))
print(f"Deleted {deleted} inactive sessions.")

Do not delete individual entries with a time-to-live index or a delete_many() filter on the entry time. Removing the oldest entries of a session that is still in use leaves a transcript that your application cannot resume.

Additional Resources

To learn more about the MongoDB session store adapter and the Claude Agent SDK, see the following resources:

• Claude Agent SDK session storage documentation

• MongoDB session store adapter repository

Visit the repository to stay updated on the adapter and consider starring the repository.