把会话持久化到外部存储

用 SessionStore 适配器把 Agent SDK 的会话记录镜像到你自己的对象存储、键值存储或数据库：SessionStore 接口、快速开始、编写适配器、参考实现与一致性测试、双写架构、从存储恢复、尽力而为的镜像写入、保留与支持范围。

默认情况下，SDK 把会话记录以 JSONL 文件写在本地文件系统的 ~/.claude/projects/ 下。SessionStore 适配器让你把这些记录镜像到自己的后端（如对象存储、键值存储或数据库），这样在一台主机上创建的会话可以在另一台从匹配工作目录运行的主机上恢复。使用会话存储的常见理由：

多主机部署。 无服务器函数、自动伸缩的 worker 和 CI runner 不共享文件系统；共享存储让副本能恢复彼此的会话。
持久性。 本地容器是临时的；外部存储能在重启和重新部署后存活。
合规与审计。 把记录保存在你已经管辖的存储里，使用你自己的保留规则、加密和访问控制。

`SessionStore` 接口

SessionStore 是一个带两个必需方法（append 和 load）和四个可选方法的对象。SDK 在查询期间调用 append 写入记录条目，调用 load 把它们读回以便恢复。

// 从 @anthropic-ai/claude-agent-sdk 导出：
// SessionStore、SessionKey、SessionStoreEntry、SessionSummaryEntry。
type SessionKey = {
  projectKey: string;
  sessionId: string;
  subpath?: string;
};

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

  // 可选
  listSessions?(
    projectKey: string,
  ): Promise<Array<{ sessionId: string; mtime: number }>>;
  listSessionSummaries?(projectKey: string): Promise<SessionSummaryEntry[]>;
  delete?(key: SessionKey): Promise<void>;
  listSubkeys?(key: {
    projectKey: string;
    sessionId: string;
  }): Promise<string[]>;
};

type SessionSummaryEntry = {
  sessionId: string;
  mtime: number;
  data: Record<string, unknown>;
};

# 从 claude_agent_sdk 导出：
# SessionStore、SessionKey、SessionStoreEntry、SessionSummaryEntry。
class SessionKey(TypedDict):
    project_key: str
    session_id: str
    subpath: NotRequired[str]


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

    # 可选——省略或抛出 NotImplementedError
    async def list_sessions(
        self, project_key: str
    ) -> list[SessionStoreListEntry]: ...
    async def list_session_summaries(
        self, project_key: str
    ) -> list[SessionSummaryEntry]: ...
    async def delete(self, key: SessionKey) -> None: ...
    async def list_subkeys(self, key: SessionListSubkeysKey) -> list[str]: ...


class SessionSummaryEntry(TypedDict):
    session_id: str
    mtime: int
    data: dict[str, Any]

SessionKey 定位一份记录。projectKey 是工作目录的稳定、对文件系统安全的编码，sessionId 是会话 UUID，subpath 在条目属于子智能体记录或旁路文件而不是主对话时设置。因为 projectKey 编码了工作目录，所以要从与原运行相匹配的工作目录从存储里恢复或继续。在 TypeScript 里，如果你在查询的 env 选项里把 CLAUDE_CODE_PROJECT_DIR_NAME 与 CLAUDE_CONFIG_DIR 一起设置，SDK 会改用该名字给该查询的条目及其 resume 和 continue 查找建键；因为 listSessions 和 deleteSession 这类独立辅助函数不接受 env、读取的是进程环境，所以也要在宿主进程环境里设置 CLAUDE_CONFIG_DIR 和同一个名字（需要 Agent SDK v0.3.234 及以上）。把 subpath 当作不透明的键后缀；它沿用磁盘上的布局，例如 subagents/agent-<id>；subpath 未定义时，键指向主记录。

方法	必需	何时调用
`append`	是	每批记录条目在本地写入之后；条目是对 JSON 安全的对象，在本地 JSONL 里每行一个
`load`	是	设了 `resume` 或 `continue: true` 解析出最新的存储会话时，在子进程派生之前；以及列表从 `listSessionSummaries` 回退时每会话一次；会话未知时返回 `null`
`listSessions`	否	由 `listSessions({ sessionStore })` 以及带 `continue: true` 的 `query()`/`startup()` 调用；未定义时，`continue: true` 会抛出，`listSessions({ sessionStore })` 也会抛出，除非实现了 `listSessionSummaries`
`listSessionSummaries`	否	由 `listSessions({ sessionStore })` 调用，一次读取所有会话的元数据；在 `append` 里维护这些摘要；未定义时，列表回退到 `listSessions` 加每会话一次 `load`
`delete`	否	由 `deleteSession({ sessionStore })` 调用；删除主键（没有 `subpath`）必须级联到该会话的所有子键，并同时移除该会话的摘要条目，使被删除的会话不再出现在 `listSessionSummaries` 里；未定义时，删除是空操作，适合只追加的后端
`listSubkeys`	否	恢复期间，用于发现子智能体记录；未定义时只还原主记录

在 SessionSummaryEntry 里，mtime 是旁路文件的存储写入时间，必须与 listSessions 返回的 mtime 值共享同一个时钟来源；data 是 SDK 拥有的不透明状态，要原样持久化而不解释它。通过在 append 里对每一批调用导出的 foldSessionSummary 辅助函数（Python 里是 fold_session_summary）来构建这些条目；跳过其键带 subpath 的批次，因为子智能体记录不得为主会话的摘要做贡献。fold 从不设置 mtime：要在持久化时盖上它，TypeScript 里通过 options.mtime 参数，Python 里通过覆盖返回条目上的字段。同一会话的并发 append 调用可能在旁路文件上竞争，所以要用事务、比较并交换或每会话锁来串行化读-fold-写。SDK 如何处理 load 返回的记录，见「从存储恢复」。

快速开始

SDK 随附一个用于开发和测试的 InMemorySessionStore。下面的例子在附带存储的情况下运行一次查询，从结果消息捕获会话 ID，然后在第二次 query() 调用里从存储恢复。第二次调用传同一个存储实例加 resume，所以 SDK 从存储而不是本地文件系统加载记录：

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

const store = new InMemorySessionStore();
let sessionId: string | undefined;

try {
  for await (const message of query({
    prompt: "List the TypeScript files under src/",
    options: { sessionStore: store },
  })) {
    if (message.type === "result") {
      sessionId = message.session_id;
    }
  }
} catch (error) {
  // 单次 query() 在产出错误结果之后抛出。如果失败是错误结果，
  // sessionId 已经被上面的循环捕获；连接或进程失败不产生结果消息。
  console.error(`Session ended with an error: ${error}`);
}

// 从存储恢复。智能体拥有第一次调用的完整上下文。
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

    try:
        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
    except Exception as error:
        # 单次 query() 在产出错误结果之后抛出。如果失败是错误结果，
        # session_id 已经被上面的循环捕获；连接或进程失败不产生结果消息。
        print(f"Session ended with an error: {error}")

    # 从存储恢复。智能体拥有第一次调用的完整上下文。
    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())

第二次查询打印第一次查询里那些文件的摘要，这表明智能体带着来自存储的完整上下文恢复了。

编写你自己的适配器

针对你的后端实现 append 和 load。如果想让 listSessions()、一次调用的元数据读取、deleteSession() 和子智能体恢复对该存储可用，再加上 listSessions、listSessionSummaries、delete 和 listSubkeys。传给 append 的条目类型为 SessionStoreEntry（形如 { type: string; ... } 的对象）。把它们当作不透明的、对 JSON 安全的值：按顺序持久化，并按同样的顺序从 load 返回。load 必须返回与所追加内容深度相等的条目；不要求字节级相等的序列化，所以重排对象键的后端（如二进制 JSON 列类型）没问题。

参考实现

两个 SDK 仓库都在 TypeScript 的 examples/session-stores/ 和 Python 的 examples/session_stores/ 下包含可运行的参考适配器。每种存储类型一个适配器，各自展示 append 和 load 如何映射到那类后端。它们不作为包发布：把最接近你后端类型的适配器复制到你的项目里，安装你后端的客户端并改造它。

存储类型	存储模型	示例适配器
对象存储	每次 `append()` 一个分片文件；`load()` 列出分片、排序并拼接	S3（TypeScript、Python）
键值存储	每份记录一个列表，`append()` 往里推、`load()` 按范围读取，外加会话的有序索引	Redis（TypeScript、Python）
关系数据库或文档存储	每个条目一行或一个文档，存为 JSON，按插入时分配的键排序	Postgres（TypeScript、Python）

每个适配器接受预先配置好的客户端实例，所以你控制凭据、TLS、区域和连接池。下面的例子把对象存储适配器接入 query()，然后在另一台主机上从它恢复：

import { query } from "@anthropic-ai/claude-agent-sdk";
import { S3Client } from "@aws-sdk/client-s3";
import { S3SessionStore } from "./S3SessionStore"; // 从 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);
  }
}

// 之后，可能在另一台主机上：
for await (const message of query({
  prompt: "Continue where we left off",
  options: { sessionStore: store, resume: "previous-session-id" },
})) {
  // ...
}

验证你的适配器

两个 SDK 都随附一套一致性测试，断言 append、load 和可选方法必须满足的行为契约；可选方法的测试在这些方法未实现时自动跳过。在 TypeScript 里，把示例目录里的 shared/conformance.ts 复制进你的测试套件；在 Python 里，该套件随包提供。要用 pytest 运行它（pytest 不是 SDK 依赖），先安装 pytest：

pip install pytest

然后在测试文件里把你的适配器作为无参数工厂传给该套件，run_session_store_conformance 对每个契约调用它一次来构建一个全新的存储：

import pytest
from claude_agent_sdk.testing import run_session_store_conformance


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

像这个例子那样直接传 MyRedisStore 类，在构造函数不带参数时可行。对接受预先配置客户端的适配器，改为传构造该存储的 lambda。因为这些契约复用相同的会话键，工厂返回的每个存储都必须从空存储开始，所以让 lambda 每次调用都准备隔离的后备存储，如全新的内存假对象、唯一的键前缀或新的测试数据库。

行为说明

双写架构

Claude Code 子进程总是先把每批记录条目写到本地磁盘，然后 SDK 把同一批转发给你存储的 append()，所以存储是本地记录的镜像，而不是它的替代。哪份副本比运行活得更久取决于运行如何开始：全新会话，或存储里没有该会话内容的恢复——你配置目录下的本地记录比运行活得更久，存储收到一份副本；从存储恢复的运行——本地副本在运行结束时被删除，所以存储持有唯一的持久副本。如果你不想让全新会话在本地磁盘上留下记录，在 options.env 里把 CLAUDE_CONFIG_DIR 设为临时目录；从存储恢复的运行已经会删除它的本地副本，所以不需要这样的设置。在 TypeScript 里，也要把 process.env 展开进 env，因为 env 选项替换子进程环境。如果你的应用通过配置目录里的文件登录（如 OAuth 凭据或用户 settings.json 里的 apiKeyHelper），要先把这些文件复制进临时目录，或改在 env 里设 ANTHROPIC_API_KEY，否则运行会以 Not logged in 失败。有两个选项与镜像冲突，把任一个与存储组合，SDK 在启动时就会抛出：TypeScript 里的 persistSession: false——关闭镜像所基于的本地写入（Python SDK 没有等价选项）；文件检查点（TypeScript 里的 enableFileCheckpointing，Python 里的 enable_file_checkpointing）——它把文件备份直接写到本地磁盘，SDK 不会把它们镜像到存储。

从存储恢复

当你把 resume（或 TypeScript 里的 continue: true、Python 里的 continue_conversation=True）与存储一起传入时，SDK 在派生子进程之前向存储要记录：resume——SDK 要你传入 ID 的那个会话；continue: true 或 continue_conversation=True——SDK 要存储里最新的会话。存储返回记录时，SDK 把它写进临时配置目录，用指向那里的 CLAUDE_CONFIG_DIR 运行子进程，并在运行结束时删除该目录；该运行写入的本地记录随之被删除，这就是这条路径上存储持有唯一持久副本的原因。SDK 还用你真实配置目录里的文件给临时目录播种，复制的内容因语言而异：

TypeScript：凭据、.claude.json 和你的用户 settings.json。从 settings.json 里它去掉在临时配置目录下行为不正常的键：enabledPlugins、extraKnownMarketplaces、它的别名 additionalMarketplaces，以及文件 env 块里的任何 CLAUDE_CONFIG_DIR（Agent SDK v0.3.232 之前，SDK 不去掉该别名）。在设置里配置的认证（如 apiKeyHelper）在从存储恢复时可用（Agent SDK v0.3.222 之前，TypeScript SDK 只复制凭据和 .claude.json）。
Python：只复制凭据和 .claude.json，所以通过用户 settings.json 里的 apiKeyHelper 认证的应用，从存储恢复时会以 Not logged in 失败；托管或项目设置里的 apiKeyHelper 仍然有效，因为 Claude Code 从 CLAUDE_CONFIG_DIR 不影响的位置读取那些文件。

存储里没有该会话的内容时，SDK 改在你真实配置目录下运行，结果取决于你传了哪个选项：resume——两个 SDK 都把 ID 传给子进程，它像没有存储时的 resume 那样恢复本地记录；TypeScript 里的 continue: true——SDK 开始全新会话；Python 里的 continue_conversation=True——SDK 从最新的本地会话继续。

镜像写入是尽力而为的

如果 append() 被拒绝，SDK 最多再用短暂退避重试该批两次，总共最多三次尝试。超时的调用不会被重试，因为原调用可能仍会落地。如果该批仍然失败，SDK 记录错误，向迭代器发出 { type: "system", subtype: "mirror_error" } 消息，丢弃该批并继续查询。因为被重试的批次可能重新投递已经落地的条目，所以要在你的 append() 实现里按 entry.uuid 去重。存储故障不会打断智能体，因为子进程先写本地。如果需要发现存储数据丢失，监控 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 }) 要求适配器实现 listSubkeys；getSubagentMessages({ sessionStore }) 在可用时使用它，未定义时回退到直接的子路径。恢复也会调用 listSubkeys 来还原子智能体文件；没有它，只物化主记录。

保留

SDK 从不自己从你的存储里删除。保留是适配器的责任：按你的合规要求，使用后端的过期或生命周期机制，或运行定时清理。CLAUDE_CONFIG_DIR 下的本地记录由 cleanupPeriodDays 设置按保留清扫规则独立清扫；从存储恢复的运行不留本地记录，所以对这些运行，你存储的保留是唯一的保留。

支持范围

下列 TypeScript SDK 函数接受 sessionStore 选项，并在提供它时对存储而不是本地文件系统操作：query()、startup()、listSessions()、getSessionInfo()、getSessionMessages()、renameSession()、tagSession()、deleteSession()、forkSession()、listSubagents()、getSubagentMessages()。

在 Python SDK 里，在 ClaudeAgentOptions 中设 session_store 来对存储运行 query()。其余操作各有一个以存储为参数的、基于存储的 Python 函数：list_sessions_from_store()、get_session_info_from_store()、get_session_messages_from_store()、list_subagents_from_store()、get_subagent_messages_from_store()、rename_session_via_store()、tag_session_via_store()、delete_session_via_store() 和 fork_session_via_store()。startup() 没有 Python 等价物。