使用checkpointing回滚文件更改

File checkpointing跟踪在agent会话期间通过Write、Edit和NotebookEdit工具进行的文件修改，允许您将文件回滚到任何之前的状态。想要尝试一下？跳转到交互式示例。使用checkpointing，您可以：

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

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

Checkpointing 如何工作

启用文件 checkpointing 时，SDK 会在通过 Write、Edit 或 NotebookEdit 工具修改文件之前创建文件备份。响应流中的用户消息包含一个 checkpoint UUID，您可以将其用作恢复点。 Checkpoint 与 agent 用来修改文件的这些内置工具一起工作：

工具	描述
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，然后稍后恢复会话以回滚文件。下面详细解释了每个步骤。本部分中的示例使用提示”重构身份验证模块”。在包含身份验证模块的项目中运行它们，或更改提示以命名项目中存在的文件，以便您可以观看文件更改并查看回滚如何恢复它们。

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

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

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")

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选项后（如上所示），响应流中的每个用户消息都有一个UUID，用作checkpoint。对于大多数用例，捕获第一个用户消息UUID（message.uuid）；回滚到它会将所有文件恢复到原始状态。要存储多个checkpoint并回滚到中间状态，请参阅多个恢复点。捕获会话ID（message.session_id）是可选的；只有在您想在流完成后回滚时才需要它。如果您在处理消息时立即调用rewindFiles()（如Checkpoint before risky operations中的示例所做的那样），您可以跳过捕获会话ID。

checkpoint_id = None
session_id = None

async for message in client.receive_response():
    # Capture the first user message UUID as the checkpoint
    if isinstance(message, UserMessage) and message.uuid and checkpoint_id is None:
        checkpoint_id = message.uuid
    # Capture session ID from the result message
    if isinstance(message, ResultMessage):
        session_id = message.session_id

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

for await (const message of response) {
  // Capture the first user message UUID as the checkpoint
  if (message.type === "user" && message.uuid && !checkpointId) {
    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 before risky operations。

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

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和checkpoint ID，您也可以从CLI回滚。此命令需要claude可执行文件，该文件来自安装Claude Code，不由SDK包安装。SDK为您启用checkpointing，但当您直接运行claude -p时，您必须设置CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING环境变量：

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

--rewind-files标志不会出现在claude --help输出中，但CLI接受它如上所示。

常见模式

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

Checkpoint before risky operations

此模式仅保留最新的checkpoint UUID，在每个agent轮次之前更新它。如果处理过程中出现问题，您可以立即回滚到最后的安全状态并跳出循环。运行此示例之前，请将your_revert_condition（Python）或yourRevertCondition（TypeScript）替换为您自己的检查，例如错误检测或验证失败；该占位符在示例中未定义。

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

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存储在带有元数据的数组中。会话完成后，您可以回滚到任何之前的checkpoint：

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

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

尝试一下

此完整示例创建一个小实用程序文件，让agent添加文档注释，向您显示更改，然后询问您是否想回滚。在开始之前，请确保您已安装Claude Agent SDK。

创建测试文件

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

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

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向您的实用程序文件添加doc注释，然后为您提供回滚和恢复原始文件的选项。

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(
                "\n✓ File restored! Open utils.py to verify the doc comments are gone."
            )
        else:
            print("\nKept the modified file.")


asyncio.run(main())

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("\n✓ File restored! Open utils.ts to verify the doc comments are gone.");
    } else {
      console.log("\nKept the modified file.");
    }
  }
}

main();

此示例演示了完整的checkpointing工作流：

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

运行示例

从与您的实用程序文件相同的目录运行脚本。

在运行脚本之前，在您的IDE或编辑器中打开您的实用程序文件（utils.py或utils.ts）。当agent添加doc注释时，您将看到文件实时更新，然后当您选择回滚时恢复到原始状态。

Python
TypeScript

python try_checkpointing.py

npx tsx try_checkpointing.ts

您将看到agent添加doc注释，然后出现一个提示，询问您是否想回滚。如果您选择是，文件将恢复到其原始状态。

限制

文件checkpointing有以下限制：

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

故障排除

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()一次。

“File rewinding is not enabled”错误

当您尝试在未启用checkpointing的情况下执行非交互式回滚时，会发生此错误：运行不带--rewind-files的裸claude -p，或运行SDK会话（包括已恢复的会话），其选项未启用checkpointing。SDK仅在启用了enable_file_checkpointing（Python）或enableFileCheckpointing（TypeScript）的会话执行回滚时，才在内部设置CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING环境变量；裸CLI永远不会设置它。 解决方案：对于裸CLI，在运行命令时设置环境变量：

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

对于SDK，在已恢复的会话上设置enable_file_checkpointing=True（Python）或enableFileCheckpointing: true（TypeScript），如本页的示例所示。

“ProcessTransport is not ready for writing”错误

当您在完成响应迭代后调用rewindFiles()或rewind_files()时，会发生此错误。当循环完成时，与CLI进程的连接关闭。 解决方案：使用空提示恢复会话，然后在新查询上调用rewind：

# 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

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

后续步骤

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

Agent SDK

核心概念

输入和输出

使用工具扩展

自定义行为

控制和可观测性

部署

SDK 参考

使用checkpointing回滚文件更改

Checkpointing 如何工作

实现checkpointing

常见模式

Checkpoint before risky operations

多个恢复点

尝试一下

限制

故障排除

Checkpointing选项未被识别

用户消息没有UUID

“No file checkpoint found for message”错误

“File rewinding is not enabled”错误

“ProcessTransport is not ready for writing”错误

后续步骤

​Checkpointing 如何工作

​实现checkpointing

​常见模式

​Checkpoint before risky operations

​多个恢复点

​尝试一下

​限制

​故障排除

​Checkpointing选项未被识别

​用户消息没有UUID

​“No file checkpoint found for message”错误

​“File rewinding is not enabled”错误

​“ProcessTransport is not ready for writing”错误

​后续步骤

Checkpointing 如何工作

实现checkpointing

常见模式

Checkpoint before risky operations

多个恢复点

尝试一下

限制

故障排除

Checkpointing选项未被识别

用户消息没有UUID

“No file checkpoint found for message”错误

“File rewinding is not enabled”错误

“ProcessTransport is not ready for writing”错误

后续步骤