Agent SDK 会话外部存储（SessionStore）参考

本文是对官方 Agent SDK「Persist sessions to external storage」页面的中文整理，完整与最新内容以原文为准：https://code.claude.com/docs/en/agent-sdk/session-storage

概述

默认情况下，SDK 会把会话转录（session transcript）以 JSONL 文件形式写入本地文件系统的 ~/.claude/projects/ 目录。SessionStore 适配器可以让你把这些转录镜像（mirror）到自己的后端，例如 S3、Redis 或数据库，这样在一台主机上创建的会话就可以在另一台工作目录相匹配的主机上恢复（resume）。

使用 session store 的常见原因：

• 多主机部署：Serverless 函数、自动扩缩容的 worker、CI runner 之间不共享文件系统。共享的 store 可以让不同副本互相恢复彼此的会话。
• 持久性：本地容器是短暂的（ephemeral）。基于 S3 或数据库的 store 能在重启和重新部署后存活。
• 合规与审计：把转录保存在你已经治理的存储中，使用自己的保留规则、加密方式和访问控制。

SessionStore 接口

SessionStore 是一个对象，包含两个必需方法 append 和 load，以及四个可选方法。SDK 在查询过程中调用 append 写入转录条目，在恢复会话时调用 load 读回这些条目。

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

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

# Exported from claude_agent_sdk as
# SessionStore, SessionKey, SessionStoreEntry, SessionSummaryEntry.

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 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]

如果在 query 的 env 里于 CLAUDE_CONFIG_DIR 之外还设置了 CLAUDE_CODE_PROJECT_DIR_NAME,TypeScript SDK 会改用该名称为这次 query 的条目以及 resume/continue 查找建键（需要 Agent SDK v0.3.234 或更高版本）。

SessionKey 用于定位一份转录。projectKey 是对工作目录的稳定、文件系统安全的编码，sessionId 是会话 UUID,subpath 仅在该条目属于子代理（subagent）转录或旁路（sidecar）文件、而非主对话时才会被设置。由于 projectKey 编码了工作目录，从 store 恢复或继续会话时，运行所在的工作目录必须与原始运行时匹配。请将 subpath 视为不透明的 key 后缀；它遵循磁盘上的目录结构，例如 subagents/agent-<id>。当 subpath 为 undefined 时，该 key 指向主转录。

在 SessionSummaryEntry 中，mtime 是旁路文件（sidecar）的存储写入时间，必须与 listSessions 返回的 mtime 值共享同一时钟源。data 是不透明的、由 SDK 自身管理的状态，应原样持久化，不要解析它。

构建这些条目时，应在 append 内部对每一批调用导出的 foldSessionSummary 辅助函数（Python 中为 fold_session_summary）。要跳过带 subpath 的批次；子代理转录不应计入主会话的摘要。该 fold 函数从不设置 mtime：需要你在持久化时打上时间戳——在 TypeScript 中通过 options.mtime 参数传入，在 Python 中则在返回的 entry 上覆写该字段。同一会话的并发 append 调用可能在旁路文件上产生竞态，因此需要用事务、compare-and-swap 或按会话加锁的方式序列化「读取-合并-写入」过程；fold 本身是纯函数。

关于 SDK 如何处理 load 返回的转录内容，见下文「从存储恢复」。

快速开始

SDK 内置了 InMemorySessionStore，用于开发和测试。下面的示例先在附加了 store 的情况下运行一次查询，从结果消息中捕获 session ID，然后在第二次 query() 调用中从 store 恢复会话。第二次调用传入了同一个 store 实例以及 resume，因此 SDK 会从 store（而非本地文件系统）加载转录：

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) {
  // A single-shot query() throws after yielding an error result. If the
  // failure was an error result, sessionId was already captured by the loop
  // above; connection or process failures yield no result message.
  console.error(`Session ended with an error: ${error}`);
}

// 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
    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:
        # A single-shot query() raises after yielding an error result. If the
        # failure was an error result, session_id was already captured by the
        # loop above; connection or process failures yield no result message.
        print(f"Session ended with an error: {error}")

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

第二次查询会打印出第一次查询涉及的文件的摘要，说明 agent 从 store 中恢复了完整上下文。

编写自己的适配器

针对你自己的后端实现 append 和 load。若希望 listSessions()、一次性元数据读取、deleteSession()、子代理恢复能正常工作，再补充实现 listSessions、listSessionSummaries、delete、listSubkeys。

传给 append 的条目类型是 SessionStoreEntry（即 { type: string; ... } 形式的对象）。应将其视为不透明的 JSON 安全值：按顺序持久化，并在 load 中按相同顺序返回。load 返回的条目必须与写入时的条目深度相等（deep-equal）；不要求字节级序列化一致，因此像 Postgres jsonb 这类会重排对象键顺序的后端也是可以的。

参考实现

TypeScript SDK 仓库在 examples/session-stores/ 下提供了可运行的 S3、Redis、Postgres 参考适配器。它们没有发布到 npm；需要把对应的 src/ 文件拷贝到你的项目中，并安装相应的后端客户端库。

每个适配器都接收一个预先配置好的客户端实例，因此凭证、TLS、region、连接池均由你自己控制。以 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 都提供了一致性测试套件（conformance suite），用于断言 append、load 以及可选方法必须满足的行为契约。针对可选方法的测试在未实现对应方法时会自动跳过。

在 TypeScript 中，把示例目录下的 shared/conformance.ts 拷贝到你的测试套件中。在 Python 中，该套件随包一同发布。要用 pytest 运行（pytest 不是 SDK 的依赖项），需先安装：

pip install pytest

然后在测试文件中，把你的适配器以零参数工厂函数的形式传给该套件；run_session_store_conformance 会对每个契约调用一次该工厂，以构建一个全新的 store:

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 来构造 store。由于各个契约会复用相同的 session key，工厂函数每次返回的 store 都必须以空存储状态开始，因此应让该 lambda 为每次调用提供隔离的后备存储，例如全新的内存态 fake、唯一的 key 前缀，或一个新的测试数据库。

行为说明

双写架构

Claude Code 子进程总是先把每批转录条目写入本地磁盘，SDK 随后再把同一批条目转发给你的 store 的 append()，因此 store 是本地转录的镜像，而非替代品。哪一份副本能在运行结束后存活，取决于运行是如何启动的：

• 全新会话，或者 store 中尚无对应会话数据的 resume：你的配置目录下的本地转录会在运行结束后存活，store 只是收到一份副本。
• 从 store 恢复的运行：本地副本会在运行结束时被删除，因此 store 中保存的是唯一的持久副本。

如果你不希望全新会话在本地磁盘留下转录，可以在 options.env 中把 CLAUDE_CONFIG_DIR 设为一个临时目录。从 store 恢复的运行本身就会删除本地副本，因此不需要这个设置。在 TypeScript 中，还应把 process.env 一并展开到 env 里，因为 env 选项 会替换子进程的整个环境变量。

如果你的应用通过配置目录下的文件（例如 OAuth 凭证，或用户 settings.json 中的 apiKeyHelper）进行登录，需要先把这些文件拷贝到临时目录中，或者改为在 env 中设置 ANTHROPIC_API_KEY。否则运行会以 Not logged in 失败。

有两个选项与镜像机制冲突，若与 store 同时使用，SDK 会在启动时抛出异常：

• persistSession: false（TypeScript）：关闭了镜像所依赖的本地写入。Python SDK 没有对应选项。
• 文件检查点（File checkpointing），即 TypeScript 中的 enableFileCheckpointing 或 Python 中的 enable_file_checkpointing：会直接把文件备份写入本地磁盘，SDK 不会把这些内容镜像到 store。

从存储恢复

当你同时传入 resume，或者 TypeScript 中的 continue: true / Python 中的 continue_conversation=True，以及一个 store 时，SDK 会在生成子进程之前先向 store 请求转录：

• resume:SDK 请求你传入的 session ID 对应的会话。
• continue: true 或 continue_conversation=True:SDK 请求 store 中最新的会话。

当 store 返回转录后，SDK 会把它写入一个临时配置目录，以 CLAUDE_CONFIG_DIR 指向该目录来运行子进程，并在运行结束时删除该目录。该次运行写入的本地转录也随之被删除，这就是为什么在这条路径下 store 是唯一持久副本的原因。

SDK 还会用你真实配置目录下的一些文件来预置该临时目录，不同语言拷贝的内容有所不同：

• TypeScript：凭证、.claude.json，以及你的用户 settings.json。从 settings.json 中，SDK 会剔除那些在临时配置目录下会出问题的键：enabledPlugins、extraKnownMarketplaces，以及它的 additionalMarketplaces 别名，还有文件 env 块中的任何 CLAUDE_CONFIG_DIR。在 Agent SDK v0.3.232 之前，SDK 不会剔除该别名。配置在 settings 中的认证方式，例如 apiKeyHelper，在从 store 恢复时依然有效。在 Agent SDK v0.3.222 之前，TypeScript SDK 只会拷贝凭证和 .claude.json。
• Python：仅拷贝凭证和 .claude.json，因此若应用通过用户 settings.json 中的 apiKeyHelper 进行认证，从 store 恢复时会以 Not logged in 失败。配置在托管（managed）或项目（project）级 settings 中的 apiKeyHelper 仍然有效，因为 Claude Code 会从不受 CLAUDE_CONFIG_DIR 影响的位置读取这些文件。

当 store 中没有对应会话的数据时，SDK 会改为在你真实的配置目录下运行，具体结果取决于你传入的选项：

• resume：两种 SDK 都会把该 ID 透传给子进程，子进程会像不带 store 时那样，直接从本地转录恢复。
• continue: true(TypeScript):SDK 会开始一个全新的会话。
• continue_conversation=True(Python):SDK 会从最新的本地会话继续。

镜像写入是尽力而为的（best-effort）

如果 append() 被拒绝（rejects），SDK 会以较短的退避间隔重试该批次最多两次，总共最多尝试三次。若某次调用超时，则不会重试，因为原始调用可能仍会成功落地。如果该批次仍然失败，SDK 会记录错误、向消息迭代器中发出一条 { type: "system", subtype: "mirror_error" } 消息、丢弃该批次，并继续本次查询。由于重试的批次可能会重新投递已经落地过的条目，你应在自己的 append() 实现中按 entry.uuid 去重。

store 故障不会中断 agent，因为子进程会先在本地写入。如果你需要检测 store 端的数据丢失，应监控 mirror_error。在从 store 恢复的运行中，被丢弃的批次一旦该次运行结束，就没有任何存活的副本了。

getSessionMessages 返回压缩后的消息链

getSessionMessages({ sessionStore }) 返回的是 agent 在恢复时会看到的、经过链接的消息链。自动压缩（auto-compaction）之后，更早的对话轮次会被替换为一段摘要，因此某个会话即使 store 中保存了 503 条原始条目，getSessionMessages 也可能只返回 18 条消息。若需要完整的原始历史，包括压缩前的对话轮次和元数据条目，应直接调用 store.load(key)。

forkSession 不是字节级拷贝

forkSession({ sessionStore }) 会读取源会话的条目，重写每一条中的 sessionId 字段并重新映射消息 UUID，然后把转换后的条目以一个新的 key 追加写入。适配器层面的复制或者类似 CopyObject 的捷径操作，会产生一份仍然引用旧 session ID 的转录，因此 SDK 不会使用这类捷径。

子代理转录

子代理（subagent）转录会以 subpath: "subagents/agent-<id>" 的形式被镜像。listSubagents({ sessionStore }) 要求适配器实现 listSubkeys;getSubagentMessages({ sessionStore }) 在该方法可用时会使用它，若未实现则回退到直接访问对应的 subpath。恢复过程也会调用 listSubkeys 来还原子代理文件；若未实现该方法，只有主转录会被还原。

保留策略（Retention）

SDK 自身不会主动从你的 store 中删除数据。数据保留是适配器自己的职责：请根据你的合规要求自行实现 TTL、S3 生命周期策略，或定期清理任务。

CLAUDE_CONFIG_DIR 下的本地转录会由 cleanupPeriodDays 设置独立清理，遵循保留清理规则。从 store 恢复的运行不会留下本地转录，因此对这类运行而言，你的 store 的保留策略就是唯一的保留策略。

支持范围

以下 TypeScript SDK 函数都接受 sessionStore 选项；当提供该选项时，它们会针对 store 而非本地文件系统进行操作：

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

在 Python SDK 中，在 ClaudeAgentOptions 中设置 session_store，可以让 query() 针对 store 运行。其余操作各自对应一个接收 store 作为参数的、store 支持的 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 中没有对应函数。Python SDK 参考文档中记录的独立函数，例如 list_sessions()，读取的是本地会话文件。

相关资源

• Work with sessions：不使用自定义 store 时的继续（continue）、恢复（resume）、分叉（fork）会话方式
• Host the SDK：多主机环境下的部署模式
• TypeScript Options：完整的选项参考
• examples/session-stores/：可运行的 S3、Redis、Postgres 参考适配器