将 Claude Agent SDK 的会话记录（transcript）镜像到 S3、Redis 或你自己的后端，使任意主机都能恢复同一会话。实现 SessionStore 接口的 append 和 load 方法即可接入；官方提供了 S3、Redis、Postgres 参考适配器，复制对应源码并安装后端客户端后即可使用。需要跨主机共享或持久化存储时建议使用，但镜像写入为尽力交付，失败时不会重试，需要监听 mirror_error 事件。

Claude Code 会话外部存储：SessionStore 配置与适配器

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

常见使用场景：

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

SessionStore 接口

SessionStore 包含两个必需方法（append 和 load）和三个可选方法。SDK 在查询期间调用 append 写入记录条目，在恢复时调用 load 读取记录。

// 从 @anthropic-ai/claude-agent-sdk 导出：
// SessionStore, SessionKey, SessionStoreEntry.

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 }>>;
  delete?(key: SessionKey): Promise<void>;
  listSubkeys?(key: {
    projectKey: string;
    sessionId: string;
  }): Promise<string[]>;
};

# 从 claude_agent_sdk 导出：
# SessionStore, SessionKey, SessionStoreEntry.

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 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。
listSessions	否	被 listSessions({ sessionStore }) 以及设置 continue: true 的 query()/startup() 调用。如果未实现，这些调用会抛出错误。
delete	否	被 deleteSession({ sessionStore }) 调用。删除主键（无 subpath）必须级联删除该会话的所有子键。如果未实现，删除为无操作，适用于仅追加的后端。
listSubkeys	否	在恢复期间发现子代理记录。如果未实现，只恢复主记录。

Quickstart

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;
  }
}

// 从存储恢复。代理拥有首次调用的完整上下文。
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

    # 从存储恢复。代理拥有首次调用的完整上下文。
    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、delete 和 listSubkeys。

传递给 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() 列出、排序并合并。
RedisSessionStore	ioredis	每个记录一个 RPUSH/LRANGE 列表，外加一个有序集合会话索引。
PostgresSessionStore	pg	每行一个条目存放在 jsonb 表中，按 BIGSERIAL 排序。

每个适配器接收一个预配置的客户端实例，因此你可以控制凭证、TLS、区域和连接池。例如 S3：

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 中，该套件随包提供：

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()。如果你希望本地副本是临时的，可以将 CLAUDE_CONFIG_DIR 指向 options.env 中的临时目录。由于镜像依赖本地写入，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 }) 要求适配器实现 listSubkeys；getSubagentMessages({ sessionStore }) 在可用时使用它，但如果未定义则回退到直接子路径。恢复时也会调用 listSubkeys 来还原子代理文件；如果没有它，只会物化主记录。

保留

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

支持的函数

以下 SDK 函数接受 sessionStore 选项，并在提供该选项时对存储而非本地文件系统进行操作：

• query()
• startup()
• listSessions()
• getSessionInfo()
• getSessionMessages()
• renameSession()
• tagSession()
• deleteSession()
• forkSession()
• listSubagents()
• getSubagentMessages()

相关资源

• 使用会话: 续用、恢复和分叉，无需自定义存储
• 托管 SDK: 多主机环境的部署模式
• TypeScript Options: 完整选项参考
• examples/session-stores/: 可运行的 S3、Redis 和 Postgres 参考适配器

常见问题

如何设置 S3 适配器？需要安装哪些依赖？

复制 examples/session-stores/s3/S3SessionStore.ts 到你的项目中，安装 @aws-sdk/client-s3，然后创建 S3Client 实例并传入 bucket 和 prefix。参考示例代码即可快速接入。

是否必须实现所有可选方法？

不需要。只实现 append 和 load 即可工作。但如果省略 listSessions，则 listSessions({ sessionStore }) 和带 continue: true 的 query() 会抛出错误；省略 listSubkeys 会导致子代理记录无法在恢复时被还原。

mirror_error 出现时会话会受影响吗？

不会。镜像写入失败只会记录错误并产生一条系统消息，本地磁盘上的记录是完整的。代理继续运行，数据不会丢失。但失败的批量不会重试，因此如果需要确保存储侧数据完整，需要主动监控 mirror_error。