import asynciofrom 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())
1
啟用 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_id = Nonesession_id = Noneasync 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
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>
import asynciofrom claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, UserMessageasync 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 breakasyncio.run(main())
如果 Claude 在多個轉向中進行變更,您可能想要回溯到特定點而不是一直回到開始。例如,如果 Claude 在第一個轉向中重構檔案,在第二個轉向中新增測試,您可能想要保留重構但撤銷測試。此模式將所有 checkpoint UUID 儲存在具有中繼資料的陣列中。工作階段完成後,您可以回溯到任何先前的 checkpoint:
import asynciofrom dataclasses import dataclassfrom datetime import datetimefrom claude_agent_sdk import ( ClaudeSDKClient, ClaudeAgentOptions, UserMessage, ResultMessage,)# Store checkpoint metadata for better tracking@dataclassclass Checkpoint: id: str description: str timestamp: datetimeasync 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())
def add(a, b): return a + bdef subtract(a, b): return a - bdef multiply(a, b): return a * bdef divide(a, b): if b == 0: raise ValueError("Cannot divide by zero") return a / b
2
執行互動式範例
在與您的公用程式檔案相同的目錄中建立一個名為 try_checkpointing.py(Python)或 try_checkpointing.ts(TypeScript)的新檔案,並貼上以下程式碼。此指令碼要求 Claude 將文件註解新增到您的公用程式檔案,然後為您提供回溯和還原原始檔案的選項。
import asynciofrom 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())
# Resume session with empty prompt, then rewindasync 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