将MongoDB与 Claude Agent SDK 集成

您可以将MongoDB与 Claude Agent SDK 集成，以跨主机保持和恢复代理会话。 Claude Agent SDK 是 Anthropic 的一个Python库，它可以在应用程序中将 Claude Code 作为代理运行。

Overview

claude-agent-sdk-mongodb-session-store包为应用程序中的 Claude Agent SDK 对话提供了MongoDB支持的会话存储适配器。会话是对话和代理采取的操作的记录，作为单独的条目存储。默认下，Claude Agent SDK 会将副本存储在调用代理的托管上。 MongoDB适配器还会将这些条目保存到MongoDB，因此副本会同时保留在托管和MongoDB中。然后，您可以从其他托管恢复会话。

您可以使用该适配器执行以下任务：

从不同进程或托管恢复会话
将会话历史记录保留在与应用程序数据相同的部署中
使用MongoDB工具查询和Atlas 审核会话历史记录

安装和设置

要使用适配器，必须有以下可用组件：

Python 3.10 或更高版本
运行MongoDB 7.0 或更高版本的MongoDB 部署
Anthropic API密钥

要学习；了解如何获取和设立Anthropic API密钥，请参阅 Claude Agent SDK 快速入门。会话存储不会使用也不会看到此密钥。

在终端中，运行以下命令以安装适配器及其依赖项：

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

此命令安装适配器及其 claude-agent-sdk 和 pymongo 依赖项。

在环境变量中设置连接字符串，如以下示例所示。适配器使用您创建的客户端，因此您的凭证会保留在客户端配置中：

export MONGODB_URI="<connection-string>"

保持会话

默认下，Claude Agent SDK 仅在运行代理的托管上保留每个会话副本。以下示例创建MongoDB会话存储、初始化其索引并运行查询。当代理运行时，适配器会将会话副本保存到MongoDB 。该示例会打印结果和 SDK 分配的会话ID 。

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())

恢复会话

要恢复会话，请创建一个 ClaudeAgentOptions对象并传递存储对象和要恢复的会话的会话ID 。 SDK 在首次运行的结果消息的 session_id字段中返回会话ID 。

您可以在异步函数中使用以下代码片段来恢复会话。将 <session-id> 替换为您的会话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)

保留和操作注意事项

适配器将每个记录条目存储为文档，并且从不自行删除会话。要删除不再活动的会话，请按安排调用 delete_inactive()，例如从 Cron作业或Atlas定时触发器中：

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.")

请勿删除带有生存时间索引或针对条目时间的 delete_many()过滤的单个条目。删除仍在使用的会话中最早的条目会留下应用程序无法恢复的记录。

客户端配置

会话存储使用您的 AsyncMongoClient 的配置，并且不设置自己的选项。使用以下生产设置：

readPreference="primary"：会话恢复时读取最新条目
w="majority"：仅在大多数副本集成员确认后才确认追加
serverSelectionTimeoutMS=10_000：阻止客户端无限期等待可用服务器
timeoutMS=20_000：阻止操作超过 Claude Agent SDK 允许的 60 秒限制

要学习；了解有关读取偏好和写入关注的更多信息，请参阅读取偏好（read preference）和写关注（write concern）文档。

并发

该存储专为每个会话一个写入者而设计。避免同时运行多个写入同一会话的进程，因为额外的写入程序可能会导致追加操作失败。仅向应用程序用户授予数据库上的 readWrite角色。要学习；了解有关数据库角色的更多信息，请参阅内置角色文档。

数据安全和隐私

您的成绩单存储可能包含敏感信息。每个条目都存储用户提示、助手回复、工具输入和结果。其中可以包括文件内容、命令输出、凭证和个人数据。使用以下步骤保护您的成绩单数据：

通过 TLS 连接以保护传输中的数据。
使用专用数据库用户进行身份验证。
仅向专用用户授予数据库上的 readWrite角色。
启用静态加密。

要学习；了解有关通过加密和身份验证保护数据的更多信息，请参阅加密文档。

其他资源

要学习；了解有关MongoDB会话存储适配器和 Claude Agent SDK 的更多信息，请参阅以下资源：

访问存储库以了解适配器的最新信息，并考虑为存储库加注星标。