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

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

捕获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():
    # 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

回滚文件

要在流完成后回滚，使用空提示恢复会话，并使用您的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

如果您捕获了会话ID和checkpoint ID，您也可以从CLI回滚：

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

常见模式

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

Checkpoint before risky operations

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

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

多个恢复点

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

尝试一下

此完整示例创建一个小实用程序文件，让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

运行交互式示例

在与您的实用程序文件相同的目录中创建一个名为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())

此示例演示了完整的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与创建它们的会话相关联
仅文件内容	创建、移动或删除目录不会通过回滚撤销
本地文件	远程或网络文件不被跟踪

Troubleshooting

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进程的连接关闭。 解决方案：使用空提示恢复会话，然后在新查询上调用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

后续步骤

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

​Checkpointing如何工作

​实现checkpointing

​常见模式

​Checkpoint before risky operations

​多个恢复点

​尝试一下

​限制

​Troubleshooting

​Checkpointing选项未被识别

​用户消息没有UUID

​“No file checkpoint found for message”错误

​“ProcessTransport is not ready for writing”错误

​后续步骤

Checkpointing如何工作

实现checkpointing

常见模式

Checkpoint before risky operations

多个恢复点

尝试一下

限制

Troubleshooting

Checkpointing选项未被识别

用户消息没有UUID

“No file checkpoint found for message”错误

“ProcessTransport is not ready for writing”错误

后续步骤