English ← MyDocs

文档索引

在此处获取完整文档索引:https://code.claude.com/docs/llms.txt 使用此文件发现所有可用页面,然后再进一步探索。

将会话持久化到外部存储

将会话记录镜像到 S3、Redis 或你自己的后端,以便任何主机都可以恢复它们。

默认情况下,SDK 将会话记录写入本地文件系统上的 ~/.claude/projects/ 下的 JSONL 文件。SessionStore 适配器允许你将这些记录镜像到你自己的后端,如 S3、Redis 或数据库,以便在一个主机上创建的会话可以在另一个主机上恢复。

使用会话存储的常见原因:

  • 多主机部署。 无服务器函数、自动扩缩工作者和 CI 运行器不共享文件系统。共享存储允许任何副本恢复任何会话。
  • 持久性。 本地容器是临时的。由 S3 或数据库支持的存储可以在重启和重新部署后幸存。
  • 合规性和审计。 将记录保存在你已经管理的存储中,使用你自己的保留规则、加密和访问控制。

SessionStore 接口

SessionStore 是一个具有两个必需方法(appendload)和三个可选方法的对象。SDK 在查询期间调用 append 写入记录条目,并在恢复时调用 load 读取它们。

// Exported from @anthropic-ai/claude-agent-sdk as
// SessionStore, SessionKey, SessionStoreEntry.

type SessionKey = {
  projectKey: string;
  sessionId: string;
  subpath?: string;
};

type SessionStore = {
  // Required
  append(key: SessionKey, entries: SessionStoreEntry[]): Promise<void>;
  load(key: SessionKey): Promise<SessionStoreEntry[] | null>;

  // Optional
  listSessions?(
    projectKey: string,
  ): Promise<Array<{ sessionId: string; mtime: number }>>;
  delete?(key: SessionKey): Promise<void>;
  listSubkeys?(key: {
    projectKey: string;
    sessionId: string;
  }): Promise<string[]>;
};
# Exported from claude_agent_sdk as
# SessionStore, SessionKey, SessionStoreEntry.

class SessionKey(TypedDict):
    project_key: str
    session_id: str
    subpath: NotRequired[str]

class SessionStore(Protocol):
    # Required
    async def append(
        self, key: SessionKey, entries: list[SessionStoreEntry]
    ) -> None: ...
    async def load(self, key: SessionKey) -> list[SessionStoreEntry] | None: ...

    # Optional — omit or raise NotImplementedError
    async def list_sessions(
        self, project_key: str
    ) -> list[SessionStoreListEntry]: ...
    async def delete(self, key: SessionKey) -> None: ...
    async def list_subkeys(self, key: SessionListSubkeysKey) -> list[str]: ...

SessionKey 寻址一条记录。projectKey 是工作目录的稳定、文件系统安全编码,sessionId 是会话 UUID,当条目属于子智能体记录或附加文件而非主对话时设置 subpath。将 subpath 视为不透明的键后缀;它遵循磁盘上的布局,例如 subagents/agent-<id>。当 subpath 未定义时,键指向主记录。

方法必需调用时机
append每批记录条目在本地写入后。条目是 JSON 安全对象,本地 JSONL 中每行一个。
load在子进程生成之前调用一次,当设置了 resume 时。如果会话未知则返回 null
listSessionslistSessions({ sessionStore }) 和带 continue: truequery()/startup() 调用。如果未定义,这些调用会抛出异常。
deletedeleteSession({ sessionStore }) 调用。删除主键(无 subpath)必须级联删除该会话的所有子键。如果未定义,删除为空操作,适用于仅追加后端。
listSubkeys在恢复期间,用于发现子智能体记录。如果未定义,仅恢复主记录。

快速开始

SDK 附带一个用于开发和测试的 InMemorySessionStore。以下示例使用附加的存储运行查询,从结果消息中捕获会话 ID,然后在第二次 query() 调用中从存储恢复。第二次调用传递相同的存储实例加上 resume,因此 SDK 从存储而不是本地文件系统加载记录:

import { query, InMemorySessionStore } from "@anthropic-ai/claude-agent-sdk";

const store = new InMemorySessionStore();

let sessionId: string | undefined;
for await (const message of query({
  prompt: "List the TypeScript files under src/",
  options: { sessionStore: store },
})) {
  if (message.type === "result") {
    sessionId = message.session_id;
  }
}

// Resume from the store. The agent has full context from the first call.
for await (const message of query({
  prompt: "Summarize what those files do",
  options: { sessionStore: store, resume: sessionId },
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}
import asyncio
from claude_agent_sdk import (
    ClaudeAgentOptions,
    InMemorySessionStore,
    ResultMessage,
    query,
)

store = InMemorySessionStore()


async def main():
    session_id = None
    async for message in query(
        prompt="List the Python files under src/",
        options=ClaudeAgentOptions(session_store=store),
    ):
        if isinstance(message, ResultMessage):
            session_id = message.session_id

    # Resume from the store. The agent has full context from the first call.
    async for message in query(
        prompt="Summarize what those files do",
        options=ClaudeAgentOptions(session_store=store, resume=session_id),
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)


asyncio.run(main())

编写你自己的适配器

针对你的后端实现 appendload。如果你想让 listSessions()deleteSession() 和子智能体恢复对存储有效,请添加 listSessionsdeletelistSubkeys

传递给 append 的条目类型为 SessionStoreEntry(一个 { type: string; ... } 对象)。将它们视为不透明的 JSON 安全值:按顺序持久化并从 load 中按相同顺序返回。load 必须返回与追加内容深度相等的条目;不需要字节相等的序列化,因此像 Postgres jsonb 这样重新排序对象键的后端是可以的。

参考实现

TypeScript SDK 仓库在 examples/session-stores/ 下包含可运行的 S3、Redis 和 Postgres 参考适配器。它们没有发布到 npm;将你需要的 src/ 文件复制到你的项目中,并安装相应的后端客户端。

适配器后端客户端存储模型
S3SessionStore@aws-sdk/client-s3每次 append() 一个 JSONL 分片文件;load() 列出、排序并拼接。
RedisSessionStoreioredis每条记录一个 RPUSH/LRANGE 列表,加上排序集合会话索引。
PostgresSessionStorepgjsonb 表中每条条目一行,按 BIGSERIAL 排序。

每个适配器接受预配置的客户端实例,因此你可以控制凭据、TLS、区域和连接池。例如,使用 S3:

import { query } from "@anthropic-ai/claude-agent-sdk";
import { S3Client } from "@aws-sdk/client-s3";
import { S3SessionStore } from "./S3SessionStore"; // copied from examples/session-stores/s3

const store = new S3SessionStore({
  bucket: "my-claude-sessions",
  prefix: "transcripts",
  client: new S3Client({ region: "us-east-1" }),
});

for await (const message of query({
  prompt: "Hello!",
  options: { sessionStore: store },
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

// Later, possibly on a different host:
for await (const message of query({
  prompt: "Continue where we left off",
  options: { sessionStore: store, resume: "previous-session-id" },
})) {
  // ...
}

验证你的适配器

两个 SDK 都附带一个一致性测试套件,用于断言 appendload 和可选方法必须满足的行为契约。当可选方法未实现时,相关的测试会自动跳过。

在 TypeScript 中,将 shared/conformance.ts 从示例目录复制到你的测试套件中。在 Python 中,测试套件随包一起发布:

import pytest
from claude_agent_sdk.testing import run_session_store_conformance


@pytest.mark.asyncio
async def test_my_store_conformance():
    await run_session_store_conformance(MyRedisStore)

行为说明

双写架构

存储是镜像,而非替代。Claude Code 子进程始终先写入本地磁盘;然后 SDK 将每批数据转发到 append()。如果你想让本地副本是临时的,请在 options.env 中将 CLAUDE_CONFIG_DIR 指向临时目录。由于镜像依赖于本地写入,sessionStore 不能与 persistSession: false 组合使用;如果同时设置两者,SDK 会抛出异常。它也不能与 enableFileCheckpointing 组合使用,因为文件历史备份 blob 直接写入本地磁盘,不会镜像到存储。

镜像写入是尽力而为的

如果 append() 拒绝或超时,错误会被记录,一个 { type: "system", subtype: "mirror_error" } 消息会被发送到迭代器中,查询继续进行。本地记录已经在磁盘上持久化,因此存储中断不会中断智能体或在本地丢失数据。失败的批次不会重试,因此如果你需要检测存储数据丢失,请监控 mirror_error

getSessionMessages 返回压缩后的消息链

getSessionMessages({ sessionStore }) 返回智能体在恢复时会看到的链接消息链。自动压缩后,较早的轮次会被摘要替换,因此存储中包含 503 条原始条目的会话可能从 getSessionMessages 返回 18 条消息。要获取完整的原始历史记录(包括压缩前的轮次和元数据条目),请直接调用 store.load(key)

forkSession 不是字节复制

forkSession({ sessionStore }) 读取源条目,重写每个 sessionId 字段并重新映射消息 UUID,然后将转换后的条目追加到新键下。适配器级别的复制或 CopyObject 快捷方式会产生仍然引用旧会话 ID 的记录,因此 SDK 不使用这种方式。

子智能体记录

子智能体记录镜像在 subpath: "subagents/agent-<id>" 下。listSubagents({ sessionStore }) 要求适配器实现 listSubkeysgetSubagentMessages({ sessionStore }) 在可用时使用它,但在未定义时回退到直接子路径。恢复也会调用 listSubkeys 来恢复子智能体文件;没有它,只有主记录会被物化。

保留

SDK 永远不会自行从你的存储中删除。保留是适配器的责任:根据你的合规要求实现 TTL、S3 生命周期策略或计划清理。CLAUDE_CONFIG_DIR 下的本地记录由 cleanupPeriodDays 设置独立清理。

支持的功能

以下 SDK 函数接受 sessionStore 选项,当提供该选项时,它们会对存储而非本地文件系统进行操作:

相关资源