Claude Agent SDK 文件 Checkpointing - 回滚文件改动

文件 checkpointing 跟踪在代理会话期间通过 Write、Edit 和 NotebookEdit 工具进行的文件修改，让你可以将文件回退到任何先前状态。想要尝试一下？跳转到交互式示例。

使用 checkpointing，你可以：

• 撤销不需要的更改，将文件恢复到已知的良好状态
• 探索替代方案，恢复到检查点并尝试不同的方法
• 从错误中恢复，当代理进行不正确的修改时

⚠️ 只有通过 Write、Edit 和 NotebookEdit 工具进行的更改才被跟踪。通过 Bash 命令（如 echo > file.txt 或 sed -i）进行的更改不会被检查点系统捕获。

checkpointing 的工作原理

启用文件 checkpointing 时，SDK 会在通过 Write、Edit 或 NotebookEdit 工具修改文件之前创建文件备份。响应流中的用户消息包括一个 checkpoint UUID，你可以将其用作恢复点。

Checkpoint 与代理用于修改文件的这些内置工具一起工作：

工具	描述
Write	创建新文件或用新内容覆盖现有文件
Edit	对现有文件的特定部分进行有针对性的编辑
NotebookEdit	修改 Jupyter notebook（.ipynb 文件）中的单元格

ℹ️ 文件回退将磁盘上的文件恢复到先前状态。它不会回退对话本身。调用 rewindFiles()（TypeScript）或 rewind_files()（Python）后，对话历史和上下文保持不变。

checkpoint 系统跟踪：

• 在会话期间创建的文件
• 在会话期间修改的文件
• 修改文件的原始内容

当你回退到 checkpoint 时，创建的文件被删除，修改的文件恢复到该时间点的内容。

实现 checkpointing

要使用文件 checkpointing，在你的选项中启用它，从响应流捕获 checkpoint UUID，然后在需要恢复时调用 rewindFiles()（TypeScript）或 rewind_files()（Python）。

以下示例显示了完整的流程：启用 checkpointing，捕获 checkpoint UUID 和会话 ID 从响应流中，然后稍后恢复会话以回退文件。每个步骤将在下面详细解释。

Python

import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage,
    ResultMessage,
)


async def main():
    # Step 1: Enable checkpointing
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",  # Auto-accept file edits without prompting
        extra_args={
            "replay-user-messages": None
        },  # Required to receive checkpoint UUIDs in the response stream
    )

    checkpoint_id = None
    session_id = None

    # Run the query and capture checkpoint UUID and session ID
    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        # Step 2: Capture checkpoint UUID from the first user message
        async for message in client.receive_response():
            if isinstance(message, UserMessage) and message.uuid and not checkpoint_id:
                checkpoint_id = message.uuid
            if isinstance(message, ResultMessage) and not session_id:
                session_id = message.session_id

    # Step 3: Later, rewind by resuming the session with an empty prompt
    if checkpoint_id and session_id:
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # Empty prompt to open the connection
            async for message in client.receive_response():
                await client.rewind_files(checkpoint_id)
                break
        print(f"Rewound to checkpoint: {checkpoint_id}")


asyncio.run(main())

TypeScript

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

async function main() {
  // Step 1: Enable checkpointing
  const opts = {
    enableFileCheckpointing: true,
    permissionMode: "acceptEdits" as const, // Auto-accept file edits without prompting
    extraArgs: { "replay-user-messages": null } // Required to receive checkpoint UUIDs in the response stream
  };

  const response = query({
    prompt: "Refactor the authentication module",
    options: opts
  });

  let checkpointId: string | undefined;
  let sessionId: string | undefined;

  // Step 2: Capture checkpoint UUID from the first user message
  for await (const message of response) {
    if (message.type === "user" && message.uuid && !checkpointId) {
      checkpointId = message.uuid;
    }
    if ("session_id" in message && !sessionId) {
      sessionId = message.session_id;
    }
  }

  // Step 3: Later, rewind by resuming the session with an empty prompt
  if (checkpointId && sessionId) {
    const rewindQuery = query({
      prompt: "", // Empty prompt to open the connection
      options: { ...opts, resume: sessionId }
    });

    for await (const msg of rewindQuery) {
      await rewindQuery.rewindFiles(checkpointId);
      break;
    }
    console.log(`Rewound to checkpoint: ${checkpointId}`);
  }
}

main();

步骤一：启用 checkpointing

配置你的 SDK 选项以启用 checkpointing 并接收 checkpoint UUID：

选项	Python	TypeScript	描述
启用 checkpointing	enable_file_checkpointing=True	enableFileCheckpointing: true	跟踪文件更改以进行回退
接收 checkpoint UUID	extra_args={"replay-user-messages": None}	extraArgs: { 'replay-user-messages': null }	必须以获取流中的用户消息 UUID

Python

options = ClaudeAgentOptions(
    enable_file_checkpointing=True,
    permission_mode="acceptEdits",
    extra_args={"replay-user-messages": None},
)

async with ClaudeSDKClient(options) as client:
    await client.query("Refactor the authentication module")

TypeScript

const response = query({
  prompt: "Refactor the authentication module",
  options: {
    enableFileCheckpointing: true,
    permissionMode: "acceptEdits" as const,
    extraArgs: { "replay-user-messages": null }
  }
});

步骤二：捕获 checkpoint UUID 和会话 ID

设置 replay-user-messages 选项后（如上所示），响应流中的每个用户消息都有一个作为 checkpoint 的 UUID。

对于大多数用例，捕获第一个用户消息 UUID（message.uuid）；回退到它会将所有文件恢复到其原始状态。要存储多个检查点并回退到中间状态，请参阅多个恢复点。

捕获会话 ID（message.session_id）是可选的；只有在流完成后想要稍后回退时才需要它。如果你在仍在处理消息时立即调用 rewindFiles()（如风险操作前的 checkpoint中的示例所示），你可以跳过捕获会话 ID。

Python

checkpoint_id = None
session_id = None

async for message in client.receive_response():
    # Update checkpoint on each user message (keeps the latest)
    if isinstance(message, UserMessage) and message.uuid:
        checkpoint_id = message.uuid
    # Capture session ID from the result message
    if isinstance(message, ResultMessage):
        session_id = message.session_id

TypeScript

let checkpointId: string | undefined;
let sessionId: string | undefined;

for await (const message of response) {
  // Update checkpoint on each user message (keeps the latest)
  if (message.type === "user" && message.uuid) {
    checkpointId = message.uuid;
  }
  // Capture session ID from any message that has it
  if ("session_id" in message) {
    sessionId = message.session_id;
  }
}

步骤三：回退文件

要在流完成后回退，使用空提示恢复会话并使用你的 checkpoint UUID 调用 rewind_files()（Python）或 rewindFiles()（TypeScript）。你也可以在流期间回退；有关该模式，请参阅风险操作前的 checkpoint。

Python

async with ClaudeSDKClient(
    ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
) as client:
    await client.query("")  # Empty prompt to open the connection
    async for message in client.receive_response():
        await client.rewind_files(checkpoint_id)
        break

TypeScript

const rewindQuery = query({
  prompt: "", // Empty prompt to open the connection
  options: { ...opts, resume: sessionId }
});

for await (const msg of rewindQuery) {
  await rewindQuery.rewindFiles(checkpointId);
  break;
}

如果你捕获了会话 ID 和检查点 ID，你也可以从 CLI 回退：

claude -p --resume <session-id> --rewind-files <checkpoint-uuid>

常见模式

这些模式展示了根据你的用例捕获和使用 checkpoint UUID 的不同方法。

风险操作前的 checkpoint

此模式仅保留最近的 checkpoint UUID，在每个代理回合之前更新它。如果在处理过程中出现问题，你可以立即回退到最后一个安全状态并跳出循环。

Python

import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, UserMessage


async def main():
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        extra_args={"replay-user-messages": None},
    )

    safe_checkpoint = None

    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        async for message in client.receive_response():
            # Update checkpoint before each agent turn starts
            # This overwrites the previous checkpoint. Only keep the latest
            if isinstance(message, UserMessage) and message.uuid:
                safe_checkpoint = message.uuid

            # Decide when to revert based on your own logic
            # For example: error detection, validation failure, or user input
            if your_revert_condition and safe_checkpoint:
                await client.rewind_files(safe_checkpoint)
                # Exit the loop after rewinding, files are restored
                break


asyncio.run(main())

TypeScript

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

async function main() {
  const response = query({
    prompt: "Refactor the authentication module",
    options: {
      enableFileCheckpointing: true,
      permissionMode: "acceptEdits" as const,
      extraArgs: { "replay-user-messages": null }
    }
  });

  let safeCheckpoint: string | undefined;

  for await (const message of response) {
    // Update checkpoint before each agent turn starts
    // This overwrites the previous checkpoint. Only keep the latest
    if (message.type === "user" && message.uuid) {
      safeCheckpoint = message.uuid;
    }

    // Decide when to revert based on your own logic
    // For example: error detection, validation failure, or user input
    if (yourRevertCondition && safeCheckpoint) {
      await response.rewindFiles(safeCheckpoint);
      // Exit the loop after rewinding, files are restored
      break;
    }
  }
}

main();

多个恢复点

如果 Claude 跨多个回合进行更改，你可能希望回退到特定点而不是一直回退。例如，如果 Claude 在第一回合重构文件并在第二回合添加测试，你可能希望保留重构但撤销测试。

此模式将所有 checkpoint UUID 存储在带有元数据的数组中。会话完成后，你可以回退到任何先前的检查点：

Python

import asyncio
from dataclasses import dataclass
from datetime import datetime
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage,
    ResultMessage,
)


# Store checkpoint metadata for better tracking
@dataclass
class Checkpoint:
    id: str
    description: str
    timestamp: datetime


async def main():
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        extra_args={"replay-user-messages": None},
    )

    checkpoints = []
    session_id = None

    async with ClaudeSDKClient(options) as client:
        await client.query("Refactor the authentication module")

        async for message in client.receive_response():
            if isinstance(message, UserMessage) and message.uuid:
                checkpoints.append(
                    Checkpoint(
                        id=message.uuid,
                        description=f"After turn {len(checkpoints) + 1}",
                        timestamp=datetime.now(),
                    )
                )
            if isinstance(message, ResultMessage) and not session_id:
                session_id = message.session_id

    # Later: rewind to any checkpoint by resuming the session
    if checkpoints and session_id:
        target = checkpoints[0]  # Pick any checkpoint
        async with ClaudeSDKClient(
            ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
        ) as client:
            await client.query("")  # Empty prompt to open the connection
            async for message in client.receive_response():
                await client.rewind_files(target.id)
                break
        print(f"Rewound to: {target.description}")


asyncio.run(main())

TypeScript

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

// Store checkpoint metadata for better tracking
interface Checkpoint {
  id: string;
  description: string;
  timestamp: Date;
}

async function main() {
  const opts = {
    enableFileCheckpointing: true,
    permissionMode: "acceptEdits" as const,
    extraArgs: { "replay-user-messages": null }
  };

  const response = query({
    prompt: "Refactor the authentication module",
    options: opts
  });

  const checkpoints: Checkpoint[] = [];
  let sessionId: string | undefined;

  for await (const message of response) {
    if (message.type === "user" && message.uuid) {
      checkpoints.push({
        id: message.uuid,
        description: `After turn ${checkpoints.length + 1}`,
        timestamp: new Date()
      });
    }
    if ("session_id" in message && !sessionId) {
      sessionId = message.session_id;
    }
  }

  // Later: rewind to any checkpoint by resuming the session
  if (checkpoints.length > 0 && sessionId) {
    const target = checkpoints[0]; // Pick any checkpoint
    const rewindQuery = query({
      prompt: "", // Empty prompt to open the connection
      options: { ...opts, resume: sessionId }
    });

    for await (const msg of rewindQuery) {
      await rewindQuery.rewindFiles(target.id);
      break;
    }
    console.log(`Rewound to: ${target.description}`);
  }
}

main();

试一试

这个完整的示例创建一个小的工具文件，让代理添加文档注释，向你显示更改，然后询问是否要回退。

在开始之前，请确保你已安装 Claude Agent SDK。

步骤一：创建一个测试文件

创建一个名为 utils.py（Python）或 utils.ts（TypeScript）的新文件并粘贴以下代码：

utils.py

def add(a, b):
    return a + b


def subtract(a, b):
    return a - b


def multiply(a, b):
    return a * b


def divide(a, b):
    if b == 0:
        raise ValueError("Cannot divide by zero")
    return a / b

utils.ts

export function add(a: number, b: number): number {
  return a + b;
}

export function subtract(a: number, b: number): number {
  return a - b;
}

export function multiply(a: number, b: number): number {
  return a * b;
}

export function divide(a: number, b: number): number {
  if (b === 0) {
    throw new Error("Cannot divide by zero");
  }
  return a / b;
}

步骤二：运行交互式示例

在你的工具文件所在的同一目录中创建一个名为 try_checkpointing.py（Python）或 try_checkpointing.ts（TypeScript）的新文件，并粘贴以下代码。

此脚本要求 Claude 向你的工具文件添加文档注释，然后给你回退和恢复原始的选择。

try_checkpointing.py

import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    UserMessage,
    ResultMessage,
)


async def main():
    # Configure the SDK with checkpointing enabled
    # - enable_file_checkpointing: Track file changes for rewinding
    # - permission_mode: Auto-accept file edits without prompting
    # - extra_args: Required to receive user message UUIDs in the stream
    options = ClaudeAgentOptions(
        enable_file_checkpointing=True,
        permission_mode="acceptEdits",
        extra_args={"replay-user-messages": None},
    )

    checkpoint_id = None  # Store the user message UUID for rewinding
    session_id = None  # Store the session ID for resuming

    print("Running agent to add doc comments to utils.py...\n")

    # Run the agent and capture checkpoint data from the response stream
    async with ClaudeSDKClient(options) as client:
        await client.query("Add doc comments to utils.py")

        async for message in client.receive_response():
            # Capture the first user message UUID - this is our restore point
            if isinstance(message, UserMessage) and message.uuid and not checkpoint_id:
                checkpoint_id = message.uuid
            # Capture the session ID so we can resume later
            if isinstance(message, ResultMessage):
                session_id = message.session_id

    print("Done! Open utils.py to see the added doc comments.\n")

    # Ask the user if they want to rewind the changes
    if checkpoint_id and session_id:
        response = input("Rewind to remove the doc comments? (y/n): ")

        if response.lower() == "y":
            # Resume the session with an empty prompt, then rewind
            async with ClaudeSDKClient(
                ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
            ) as client:
                await client.query("")  # Empty prompt opens the connection
                async for message in client.receive_response():
                    await client.rewind_files(checkpoint_id)  # Restore files
                    break

            print(
                "\nFile restored! Open utils.py to verify the doc comments are gone."
            )
        else:
            print("\nKept the modified file.")


asyncio.run(main())

try_checkpointing.ts

import { query } from "@anthropic-ai/claude-agent-sdk";
import * as readline from "readline";

async function main() {
  // Configure the SDK with checkpointing enabled
  // - enableFileCheckpointing: Track file changes for rewinding
  // - permissionMode: Auto-accept file edits without prompting
  // - extraArgs: Required to receive user message UUIDs in the stream
  const opts = {
    enableFileCheckpointing: true,
    permissionMode: "acceptEdits" as const,
    extraArgs: { "replay-user-messages": null }
  };

  let sessionId: string | undefined; // Store the session ID for resuming
  let checkpointId: string | undefined; // Store the user message UUID for rewinding

  console.log("Running agent to add doc comments to utils.ts...\n");

  // Run the agent and capture checkpoint data from the response stream
  const response = query({
    prompt: "Add doc comments to utils.ts",
    options: opts
  });

  for await (const message of response) {
    // Capture the first user message UUID - this is our restore point
    if (message.type === "user" && message.uuid && !checkpointId) {
      checkpointId = message.uuid;
    }
    // Capture the session ID so we can resume later
    if ("session_id" in message) {
      sessionId = message.session_id;
    }
  }

  console.log("Done! Open utils.ts to see the added doc comments.\n");

  // Ask the user if they want to rewind the changes
  if (checkpointId && sessionId) {
    const rl = readline.createInterface({
      input: process.stdin,
      output: process.stdout
    });

    const answer = await new Promise<string>((resolve) => {
      rl.question("Rewind to remove the doc comments? (y/n): ", resolve);
    });
    rl.close();

    if (answer.toLowerCase() === "y") {
      // Resume the session with an empty prompt, then rewind
      const rewindQuery = query({
        prompt: "", // Empty prompt opens the connection
        options: { ...opts, resume: sessionId }
      });

      for await (const msg of rewindQuery) {
        await rewindQuery.rewindFiles(checkpointId); // Restore files
        break;
      }

      console.log("\nFile restored! Open utils.ts to verify the doc comments are gone.");
    } else {
      console.log("\nKept the modified file.");
    }
  }
}

main();

这个示例展示了完整的 checkpointing 工作流程：

1. 启用 checkpointing：使用 enable_file_checkpointing=True 和 permission_mode="acceptEdits" 配置 SDK 以自动批准文件编辑
2. 捕获 checkpoint 数据：当代理运行时，存储第一个用户消息 UUID（你的恢复点）和会话 ID
3. 提示回退：代理完成后，检查你的工具文件以查看文档注释，然后决定是否要撤销更改
4. 恢复和回退：如果是，使用空提示恢复会话并调用 rewind_files() 以恢复原始文件

步骤三：运行示例

从你的工具文件所在的同一目录运行该脚本。

💡 在运行脚本之前，在 IDE 或编辑器中打开你的工具文件（utils.py 或 utils.ts）。当代理添加文档注释时，你将实时看到文件更新，当你选择回退时，它会恢复为原始内容。

Python

python try_checkpointing.py

TypeScript

npx tsx try_checkpointing.ts

你会看到代理添加文档注释，然后会出现一个询问是否要回退的提示。如果你选择是，文件会恢复到原始状态。

限制

文件 checkpointing 有以下限制：

限制	描述
仅 Write/Edit/NotebookEdit 工具	通过 Bash 命令进行的更改不会被跟踪
同一会话	检查点与创建它们的会话绑定
仅文件内容	创建、移动或删除目录不会通过回退撤销
本地文件	远程或网络文件不会被跟踪

故障排除

Checkpointing 选项无法识别

如果 enableFileCheckpointing 或 rewindFiles() 不可用，你可能使用的是较旧的 SDK 版本。

解决方案：更新到最新的 SDK 版本：

• Python：pip install --upgrade claude-agent-sdk
• TypeScript：npm install @anthropic-ai/claude-agent-sdk@latest

用户消息没有 UUID

如果 message.uuid 是 undefined 或缺失，你没有收到 checkpoint UUID。

原因：没有设置 replay-user-messages 选项。

解决方案：将 extra_args={"replay-user-messages": None}（Python）或 extraArgs: { 'replay-user-messages': null }（TypeScript）添加到你的选项中。

“No file checkpoint found for message” 错误

当指定的用户消息 UUID 不存在 checkpoint 数据时会发生此错误。

常见原因：

• 原始会话上未启用文件 checkpointing（enable_file_checkpointing 或 enableFileCheckpointing 未设置为 true）
• 在尝试恢复和回退之前，会话未正确完成

解决方案：确保在原始会话上设置 enable_file_checkpointing=True（Python）或 enableFileCheckpointing: true（TypeScript），然后使用示例中显示的模式：捕获第一个用户消息 UUID，完全完成会话，然后使用空提示恢复并调用 rewindFiles() 一次。

“ProcessTransport is not ready for writing” 错误

当你在完成对响应的迭代后调用 rewindFiles() 或 rewind_files() 时会发生此错误。当循环完成时，到 CLI 进程的连接会关闭。

解决方案：使用空提示恢复会话，然后在新查询上调用回退：

Python

# Resume session with empty prompt, then rewind
async with ClaudeSDKClient(
    ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
) as client:
    await client.query("")
    async for message in client.receive_response():
        await client.rewind_files(checkpoint_id)
        break

TypeScript

// Resume session with empty prompt, then rewind
const rewindQuery = query({
  prompt: "",
  options: { ...opts, resume: sessionId }
});

for await (const msg of rewindQuery) {
  await rewindQuery.rewindFiles(checkpointId);
  break;
}

后续步骤

• 会话：了解如何恢复会话，这是流完成后回退所必需的。涵盖会话 ID、恢复对话和会话分支。
• 权限：配置 Claude 可以使用哪些工具以及如何批准文件修改。如果你希望对何时进行编辑有更多控制，这很有用。
• TypeScript SDK 参考：完整的 API 参考，包括 query() 的所有选项和 rewindFiles() 方法。
• Python SDK 参考：完整的 API 参考，包括 ClaudeAgentOptions 的所有选项和 rewind_files() 方法。

---

本文翻译自 Anthropic Claude Code 官方文档，最近一次同步：2025-05-01。