Ripristina le modifiche ai file con checkpointing
Traccia le modifiche ai file durante le sessioni dell'agente e ripristina i file a qualsiasi stato precedente
Il checkpointing dei file traccia le modifiche ai file effettuate tramite gli strumenti Write, Edit e NotebookEdit durante una sessione dell'agente, consentendoti di ripristinare i file a qualsiasi stato precedente. Vuoi provarlo? Vai all'esempio interattivo.
Con il checkpointing, puoi:
- Annullare le modifiche indesiderate ripristinando i file a uno stato noto e funzionante
- Esplorare alternative ripristinando a un checkpoint e provando un approccio diverso
- Recuperare da errori quando l'agente effettua modifiche non corrette
Solo le modifiche effettuate tramite gli strumenti Write, Edit e NotebookEdit vengono tracciate. Le modifiche effettuate tramite comandi Bash (come echo > file.txt o sed -i) non vengono acquisite dal sistema di checkpoint.
Come funziona il checkpointing
Quando abiliti il checkpointing dei file, l'SDK crea backup dei file prima di modificarli tramite gli strumenti Write, Edit o NotebookEdit. I messaggi dell'utente nel flusso di risposta includono un UUID di checkpoint che puoi utilizzare come punto di ripristino.
Checkpoint funziona con questi strumenti integrati che l'agente utilizza per modificare i file:
| Tool | Description |
|---|---|
| Write | Crea un nuovo file o sovrascrive un file esistente con nuovo contenuto |
| Edit | Effettua modifiche mirate a parti specifiche di un file esistente |
| NotebookEdit | Modifica le celle nei notebook Jupyter (file .ipynb) |
Il ripristino dei file ripristina i file su disco a uno stato precedente. Non ripristina la conversazione stessa. La cronologia della conversazione e il contesto rimangono intatti dopo la chiamata a rewindFiles() (TypeScript) o rewind_files() (Python).
Il sistema di checkpoint traccia:
- File creati durante la sessione
- File modificati durante la sessione
- Il contenuto originale dei file modificati
Quando ripristini a un checkpoint, i file creati vengono eliminati e i file modificati vengono ripristinati al loro contenuto in quel momento.
Implementare il checkpointing
Per utilizzare il checkpointing dei file, abilitalo nelle tue opzioni, acquisisci gli UUID di checkpoint dal flusso di risposta, quindi chiama rewindFiles() (TypeScript) o rewind_files() (Python) quando hai bisogno di ripristinare.
L'esempio seguente mostra il flusso completo: abilita il checkpointing, acquisisci l'UUID di checkpoint e l'ID di sessione dal flusso di risposta, quindi riprendi la sessione in seguito per ripristinare i file. Ogni passaggio è spiegato in dettaglio di seguito.
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();
Abilita il checkpointing
Configura le opzioni dell'SDK per abilitare il checkpointing e ricevere gli UUID di checkpoint:
| Option | Python | TypeScript | Description |
|---|---|---|---|
| Enable checkpointing | enable_file_checkpointing=True |
enableFileCheckpointing: true |
Traccia le modifiche ai file per il ripristino |
| Receive checkpoint UUIDs | extra_args={"replay-user-messages": None} |
extraArgs: { 'replay-user-messages': null } |
Obbligatorio per ottenere gli UUID dei messaggi dell'utente nel flusso |
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 }
}
});
Acquisisci l'UUID di checkpoint e l'ID di sessione
Con l'opzione replay-user-messages impostata (mostrata sopra), ogni messaggio dell'utente nel flusso di risposta ha un UUID che funge da checkpoint.
Per la maggior parte dei casi d'uso, acquisisci il primo UUID del messaggio dell'utente (message.uuid); il ripristino ad esso ripristina tutti i file al loro stato originale. Per archiviare più checkpoint e ripristinare a stati intermedi, vedi Più punti di ripristino.
L'acquisizione dell'ID di sessione (message.session_id) è facoltativa; ne hai bisogno solo se vuoi ripristinare in seguito, dopo il completamento del flusso. Se stai chiamando rewindFiles() immediatamente mentre stai ancora elaborando i messaggi (come fa l'esempio in Checkpoint prima di operazioni rischiose), puoi saltare l'acquisizione dell'ID di sessione.
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
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;
}
}
Ripristina i file
Per ripristinare dopo il completamento del flusso, riprendi la sessione con un prompt vuoto e chiama rewind_files() (Python) o rewindFiles() (TypeScript) con il tuo UUID di checkpoint. Puoi anche ripristinare durante il flusso; vedi Checkpoint prima di operazioni rischiose per quel modello.
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;
}
Se acquisisci l'ID di sessione e l'ID di checkpoint, puoi anche ripristinare dalla CLI:
claude -p --resume <session-id> --rewind-files <checkpoint-uuid>
Modelli comuni
Questi modelli mostrano diversi modi per acquisire e utilizzare gli UUID di checkpoint a seconda del tuo caso d'uso.
Checkpoint prima di operazioni rischiose
Questo modello mantiene solo l'UUID di checkpoint più recente, aggiornandolo prima di ogni turno dell'agente. Se qualcosa va storto durante l'elaborazione, puoi immediatamente ripristinare all'ultimo stato sicuro e uscire dal ciclo.
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();
Più punti di ripristino
Se Claude effettua modifiche su più turni, potresti voler ripristinare a un punto specifico piuttosto che all'inizio. Ad esempio, se Claude effettua il refactoring di un file nel turno uno e aggiunge test nel turno due, potresti voler mantenere il refactoring ma annullare i test.
Questo modello archivia tutti gli UUID di checkpoint in un array con metadati. Dopo il completamento della sessione, puoi ripristinare a qualsiasi checkpoint precedente:
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();
Prova
Questo esempio completo crea un piccolo file di utilità, fa aggiungere all'agente commenti di documentazione, ti mostra le modifiche, quindi ti chiede se vuoi ripristinare.
Prima di iniziare, assicurati di avere Claude Agent SDK installato.
Crea un file di test
Crea un nuovo file chiamato utils.py (Python) o utils.ts (TypeScript) e incolla il seguente codice:
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;
}
Esegui l'esempio interattivo
Crea un nuovo file chiamato try_checkpointing.py (Python) o try_checkpointing.ts (TypeScript) nella stessa directory del tuo file di utilità e incolla il seguente codice.
Questo script chiede a Claude di aggiungere commenti doc al tuo file di utilità, quindi ti dà la possibilità di ripristinare e ripristinare l'originale.
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();
Questo esempio dimostra il flusso di lavoro completo del checkpointing:
- Abilita il checkpointing: configura l'SDK con
enable_file_checkpointing=Trueepermission_mode="acceptEdits"per approvare automaticamente le modifiche ai file - Acquisisci i dati di checkpoint: mentre l'agente è in esecuzione, archivia il primo UUID del messaggio dell'utente (il tuo punto di ripristino) e l'ID di sessione
- Richiedi il ripristino: dopo che l'agente ha terminato, controlla il tuo file di utilità per vedere i commenti doc, quindi decidi se vuoi annullare le modifiche
- Riprendi e ripristina: se sì, riprendi la sessione con un prompt vuoto e chiama
rewind_files()per ripristinare il file originale
Esegui l'esempio
Esegui lo script dalla stessa directory del tuo file di utilità.
Apri il tuo file di utilità (utils.py o utils.ts) nel tuo IDE o editor prima di eseguire lo script. Vedrai il file aggiornarsi in tempo reale mentre l'agente aggiunge commenti doc, quindi tornare all'originale quando scegli di ripristinare.
python try_checkpointing.py
npx tsx try_checkpointing.ts
Vedrai l'agente aggiungere commenti doc, quindi un prompt che ti chiede se vuoi ripristinare. Se scegli sì, il file viene ripristinato al suo stato originale.
Limitazioni
Il checkpointing dei file ha le seguenti limitazioni:
| Limitation | Description |
|---|---|
| Solo strumenti Write/Edit/NotebookEdit | Le modifiche effettuate tramite comandi Bash non vengono tracciate |
| Stessa sessione | I checkpoint sono legati alla sessione che li ha creati |
| Solo contenuto del file | La creazione, lo spostamento o l'eliminazione di directory non vengono annullati dal ripristino |
| File locali | I file remoti o di rete non vengono tracciati |
Troubleshooting
Le opzioni di checkpointing non vengono riconosciute
Se enableFileCheckpointing o rewindFiles() non sono disponibili, potresti essere su una versione SDK più vecchia.
Soluzione: Aggiorna alla versione SDK più recente:
- Python:
pip install --upgrade claude-agent-sdk - TypeScript:
npm install @anthropic-ai/claude-agent-sdk@latest
I messaggi dell'utente non hanno UUID
Se message.uuid è undefined o mancante, non stai ricevendo gli UUID di checkpoint.
Causa: L'opzione replay-user-messages non è impostata.
Soluzione: Aggiungi extra_args={"replay-user-messages": None} (Python) o extraArgs: { 'replay-user-messages': null } (TypeScript) alle tue opzioni.
Errore "No file checkpoint found for message"
Questo errore si verifica quando i dati di checkpoint non esistono per l'UUID del messaggio dell'utente specificato.
Cause comuni:
- Il checkpointing dei file non era abilitato sulla sessione originale (
enable_file_checkpointingoenableFileCheckpointingnon era impostato sutrue) - La sessione non è stata completata correttamente prima di tentare di riprendere e ripristinare
Soluzione: Assicurati che enable_file_checkpointing=True (Python) o enableFileCheckpointing: true (TypeScript) fosse impostato sulla sessione originale, quindi utilizza il modello mostrato negli esempi: acquisisci il primo UUID del messaggio dell'utente, completa la sessione completamente, quindi riprendi con un prompt vuoto e chiama rewindFiles() una volta.
Errore "ProcessTransport is not ready for writing"
Questo errore si verifica quando chiami rewindFiles() o rewind_files() dopo aver terminato l'iterazione attraverso la risposta. La connessione al processo CLI si chiude quando il ciclo si completa.
Soluzione: Riprendi la sessione con un prompt vuoto, quindi chiama rewind sulla nuova query:
# 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;
}
Passaggi successivi
- Sessions: scopri come riprendere le sessioni, che è obbligatorio per il ripristino dopo il completamento del flusso. Copre gli ID di sessione, la ripresa delle conversazioni e il forking delle sessioni.
- Permissions: configura quali strumenti Claude può utilizzare e come vengono approvate le modifiche ai file. Utile se vuoi più controllo su quando avvengono le modifiche.
- TypeScript SDK reference: riferimento API completo incluse tutte le opzioni per
query()e il metodorewindFiles(). - Python SDK reference: riferimento API completo incluse tutte le opzioni per
ClaudeAgentOptionse il metodorewind_files().