文档索引
在此处获取完整文档索引:https://code.claude.com/docs/llms.txt 使用此文件发现所有可用页面,然后再进一步探索。
将会话持久化到外部存储
将会话记录镜像到 S3、Redis 或你自己的后端,以便任何主机都可以恢复它们。
默认情况下,SDK 将会话记录写入本地文件系统上的 ~/.claude/projects/ 下的 JSONL 文件。SessionStore 适配器允许你将这些记录镜像到你自己的后端,如 S3、Redis 或数据库,以便在一个主机上创建的会话可以在另一个主机上恢复。
使用会话存储的常见原因:
- 多主机部署。 无服务器函数、自动扩缩工作者和 CI 运行器不共享文件系统。共享存储允许任何副本恢复任何会话。
- 持久性。 本地容器是临时的。由 S3 或数据库支持的存储可以在重启和重新部署后幸存。
- 合规性和审计。 将记录保存在你已经管理的存储中,使用你自己的保留规则、加密和访问控制。
SessionStore 接口
SessionStore 是一个具有两个必需方法(append 和 load)和三个可选方法的对象。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。 |
listSessions | 否 | 由 listSessions({ sessionStore }) 和带 continue: true 的 query()/startup() 调用。如果未定义,这些调用会抛出异常。 |
delete | 否 | 由 deleteSession({ 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())
编写你自己的适配器
针对你的后端实现 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"; // 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 都附带一个一致性测试套件,用于断言 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()。如果你想让本地副本是临时的,请在 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 }) 要求适配器实现 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 参考适配器