Introduction to Error Recovery Patterns with Claude Code
When building applications powered by Claude Code, errors are inevitable. Whether it's a malformed API response, a rate limit hit, a network timeout, or an unexpected model output, your code needs to gracefully handle these situations. Error recovery patterns are structured approaches to detecting, responding to, and recovering from failures in a way that maintains application stability and user experience.
In this complete guide, we'll explore the most effective error recovery patterns specifically tailored for Claude Code workflows. You'll learn how to build resilient integrations that can withstand real-world conditions, from transient network issues to complex reasoning failures.
Why Error Recovery Matters
LLM-powered applications face a unique set of failure modes that traditional software doesn't encounter. A standard REST API either succeeds or returns a structured error, but Claude Code can produce syntactically valid responses that are semantically wrong, get stuck in tool-use loops, or exceed context windows mid-conversation. Without proper recovery patterns, these failures cascade into broken user experiences.
Common Failure Scenarios
- Transient network errors: Connection resets, DNS failures, and timeouts during API calls.
- Rate limiting: Hitting tokens-per-minute or requests-per-minute thresholds.
- Context window overflow: Conversations growing beyond the model's context limit.
- Malformed tool calls: Claude attempting to call tools with invalid or missing parameters.
- Looping behavior: The model repeating the same action without making progress.
- Unexpected output format: Responses that don't match your expected schema.
Pattern 1: Retry with Exponential Backoff
The foundational error recovery pattern for any networked application is retry with exponential backoff. This is especially important for Claude Code because the API enforces rate limits that may cause temporary failures. The key is to retry only on idempotent or transient errors, not on permanent failures like authentication errors.
import time
import random
from anthropic import Anthropic, APIError, RateLimitError, APITimeoutError
client = Anthropic()
def call_claude_with_retry(
messages,
max_retries=5,
base_delay=1.0,
max_delay=60.0
):
"""
Call Claude with exponential backoff and jitter.
Only retries on transient errors.
"""
last_error = None
for attempt in range(max_retries):
try:
response = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=4096,
messages=messages
)
return response
except RateLimitError as e:
last_error = e
# Rate limits need longer backoff
delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
print(f"Rate limited. Retrying in {delay:.1f}s (attempt {attempt + 1})")
time.sleep(delay)
except APITimeoutError as e:
last_error = e
delay = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
print(f"Timeout. Retrying in {delay:.1f}s (attempt {attempt + 1})")
time.sleep(delay)
except APIError as e:
# Check if it's a retryable error
if hasattr(e, 'status_code') and e.status_code >= 500:
last_error = e
delay = min(base_delay * (2 ** attempt), max_delay)
print(f"Server error {e.status_code}. Retrying in {delay:.1f}s")
time.sleep(delay)
else:
# Non-retryable error (4xx), raise immediately
raise
raise Exception(f"Max retries exceeded. Last error: {last_error}")
The addition of jitter (randomized delay) is critical when running multiple concurrent requests. Without jitter, retrying clients tend to synchronize their retry attempts, creating thundering herd problems that can worsen rate limiting.
Pattern 2: Circuit Breaker
When Claude's API is experiencing an extended outage, continuously retrying wastes resources and can delay recovery. The circuit breaker pattern monitors failure rates and temporarily stops sending requests when the error threshold is exceeded, giving the system time to recover.
import time
from enum import Enum
from functools import wraps
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing if service recovered
class CircuitBreaker:
def __init__(
self,
failure_threshold=5,
recovery_timeout=60,
half_open_max_calls=3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_max_calls = half_open_max_calls
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time = None
self.half_open_calls = 0
def call(self, func, *args, **kwargs):
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
else:
raise Exception("Circuit breaker is OPEN - service unavailable")
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls >= self.half_open_max_calls:
raise Exception("Circuit breaker HALF_OPEN - max test calls reached")
self.half_open_calls += 1
try:
result = func(*args, **kwargs)
self._on_success()
return result
except Exception as e:
self._on_failure()
raise
def _should_attempt_reset(self):
return (
self.last_failure_time is not None and
time.time() - self.last_failure_time >= self.recovery_timeout
)
def _on_success(self):
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.half_open_max_calls:
self.state = CircuitState.CLOSED
self.success_count = 0
def _on_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.success_count = 0
elif self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
# Usage with Claude Code
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)
def safe_claude_call(messages):
return breaker.call(
client.messages.create,
model="claude-sonnet-4-20250514",
max_tokens=4096,
messages=messages
)
Pattern 3: Fallback Chain
Sometimes the best recovery strategy is to try a different approach entirely. A fallback chain lets you degrade gracefully — for example, falling back from a larger model to a smaller, faster one, or from a tool-augmented call to a simpler direct prompt. This ensures your application remains functional even when the primary approach fails.
from anthropic import Anthropic
client = Anthropic()
class FallbackChain:
"""Chain of fallback strategies for Claude Code calls."""
def __init__(self, strategies):
self.strategies = strategies
def execute(self, prompt, context=None):
errors = []
for i, strategy in enumerate(self.strategies):
try:
result = strategy(prompt, context)
if self._is_valid_result(result):
if i > 0:
print(f"Recovered using fallback strategy {i}")
return result
else:
errors.append(f"Strategy {i}: invalid result")
except Exception as e:
errors.append(f"Strategy {i}: {str(e)}")
continue
raise Exception(f"All fallback strategies failed: {errors}")
def _is_valid_result(self, result):
"""Validate that the result is usable."""
if result is None:
return False
if hasattr(result, 'content') and len(result.content) == 0:
return False
return True
# Define fallback strategies
def strategy_primary_model(prompt, context):
"""Use the most capable model with full tools."""
return client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=8192,
tools=context.get("tools", []),
messages=[{"role": "user", "content": prompt}]
)
def strategy_smaller_model(prompt, context):
"""Fall back to a faster, smaller model."""
return client.messages.create(
model="claude-3-5-haiku-20241022",
max_tokens=4096,
messages=[{"role": "user", "content": prompt}]
)
def strategy_simplified_prompt(prompt, context):
"""Simplify the prompt and remove tool complexity."""
simplified = f"Answer concisely: {prompt}"
return client.messages.create(
model="claude-3-5-haiku-20241022",
max_tokens=2048,
messages=[{"role": "user", "content": simplified}]
)
def strategy_cached_response(prompt, context):
"""Return a cached or default response as last resort."""
cache_key = hash(prompt)
cached = response_cache.get(cache_key)
if cached:
return cached
raise Exception("No cached response available")
# Build the chain
chain = FallbackChain([
strategy_primary_model,
strategy_smaller_model,
strategy_simplified_prompt,
strategy_cached_response,
])
# Use it
result = chain.execute("Analyze this code for bugs", context={"tools": tools})
Pattern 4: Context Window Recovery
One of the most common errors in extended Claude Code sessions is context window overflow. As conversations grow longer, you'll eventually exceed the model's context limit. The recovery pattern here involves intelligently summarizing or truncating the conversation history while preserving the most important context.
class ContextWindowManager:
"""Manages conversation context to prevent overflow."""
def __init__(self, max_tokens=180000, reserve_for_response=8192):
self.max_tokens = max_tokens
self.reserve_for_response = reserve_for_response
self.effective_limit = max_tokens - reserve_for_response
def count_tokens(self, messages):
"""Estimate token count for messages."""
total = 0
for msg in messages:
# Rough estimate: ~4 chars per token
content = msg.get("content", "")
if isinstance(content, str):
total += len(content) // 4
elif isinstance(content, list):
for block in content:
if isinstance(block, dict):
total += len(str(block.get("text", ""))) // 4
return total
def recover_context(self, messages, system_prompt=None):
"""
Recover from context overflow by summarizing older messages.
Preserves the most recent messages and any system-critical context.
"""
token_count = self.count_tokens(messages)
if token_count <= self.effective_limit:
return messages # No recovery needed
print(f"Context overflow detected: {token_count} tokens. Recovering...")
# Strategy: Keep the last N messages, summarize the rest
# Preserve recent tool calls and their results
preserved_count = self._find_preservation_boundary(messages)
old_messages = messages[:preserved_count]
recent_messages = messages[preserved_count:]
# Summarize old messages
summary = self._summarize_messages(old_messages)
# Rebuild conversation with summary
recovered = [
{
"role": "user",
"content": f"[Previous conversation summary]: {summary}"
},
{
"role": "assistant",
"content": "Understood. I have the context from our previous conversation. Let's continue."
}
] + recent_messages
new_count = self.count_tokens(recovered)
print(f"Recovered context: {token_count} -> {new_count} tokens")
return recovered
def _find_preservation_boundary(self, messages):
"""Find a good boundary to split messages.
Avoid cutting in the middle of a tool-use exchange."""
# Keep at least the last 10 messages
min_preserve = 10
boundary = len(messages) - min_preserve
# Walk backwards to find a clean break (after an assistant response)
for i in range(boundary, 0, -1):
if messages[i].get("role") == "user":
return i
return boundary
def _summarize_messages(self, messages):
"""Use Claude to summarize older conversation messages."""
conversation_text = "\n".join([
f"{m['role']}: {m['content'] if isinstance(m['content'], str) else str(m['content'])}"
for m in messages
])
response = client.messages.create(
model="claude-3-5-haiku-20241022",
max_tokens=1024,
messages=[{
"role": "user",
"content": f"Summarize this conversation concisely, "
f"preserving key decisions, code context, and action items:\n\n{conversation_text}"
}]
)
return response.content[0].text
# Usage in a conversation loop
context_manager = ContextWindowManager(max_tokens=200000)
def chat_turn(messages, user_input):
messages.append({"role": "user", "content": user_input})
# Check and recover context before each call
messages = context_manager.recover_context(messages)
response = call_claude_with_retry(messages)
messages.append({"role": "assistant", "content": response.content[0].text})
return messages, response
Pattern 5: Tool Call Validation and Recovery
When Claude Code uses tools, it sometimes generates malformed tool calls — missing required parameters, wrong types, or calling non-existent tools. Rather than failing outright, you should validate tool calls and provide corrective feedback to the model, giving it a chance to fix its own mistakes.
import json
from typing import Any, Dict
class ToolCallRecovery:
"""Validates and recovers from malformed tool calls."""
def __init__(self, tool_definitions):
self.tools = {tool["name"]: tool for tool in tool_definitions}
def validate_tool_call(self, tool_name: str, tool_input: Dict) -> Dict:
"""Validate a tool call and return errors if any."""
errors = []
# Check if tool exists
if tool_name not in self.tools:
return {
"valid": False,
"errors": [f"Unknown tool: '{tool_name}'. Available tools: {list(self.tools.keys())}"]
}
tool_def = self.tools[tool_name]
schema = tool_def.get("input_schema", {})
required = schema.get("required", [])
properties = schema.get("properties", {})
# Check required fields
for field in required:
if field not in tool_input:
errors.append(f"Missing required parameter: '{field}'")
# Check types
for field, value in tool_input.items():
if field in properties:
expected_type = properties[field].get("type")
if expected_type and not self._check_type(value, expected_type):
errors.append(
f"Parameter '{field}' expected type '{expected_type}', "
f"got '{type(value).__name__}'"
)
return {"valid": len(errors) == 0, "errors": errors}
def _check_type(self, value, expected_type):
type_map = {
"string": str,
"integer": int,
"number": (int, float),
"boolean": bool,
"array": list,
"object": dict,
}
expected = type_map.get(expected_type)
if expected is None:
return True
return isinstance(value, expected) and not (expected_type == "boolean" and isinstance(value, (int, str)))
def recover_tool_call(self, messages, tool_use_block, tool_result_error):
"""
Send corrective feedback to Claude about a malformed tool call.
"""
correction_message = {
"role": "user",
"content": [
{
"type": "text",
"text": (
f"The previous tool call had errors:\n{tool_result_error}\n\n"
f"Please correct the tool call and try again. "
f"Make sure all required parameters are present and correctly typed."
)
}
]
}
# Add the correction and let Claude retry
messages.append({
"role": "assistant",
"content": [tool_use_block]
})
messages.append({
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": tool_use_block.id,
"content": f"Error: {tool_result_error}",
"is_error": True
}
]
})
messages.append(correction_message)
return messages
# Usage in a tool-use loop
def execute_with_tool_recovery(messages, max_tool_rounds=10):
recovery = ToolCallRecovery(tools)
for round_num in range(max_tool_rounds):
response = call_claude_with_retry(messages)
# Check if response contains tool calls
tool_calls = [block for block in response.content if block.type == "tool_use"]
if not tool_calls:
# No tool calls, return final response
return response
messages.append({"role": "assistant", "content": response.content})
# Process each tool call
tool_results = []
for tc in tool_calls:
validation = recovery.validate_tool_call(tc.name, tc.input)
if validation["valid"]:
# Execute the tool
try:
result = execute_tool(tc.name, tc.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": tc.id,
"content": str(result)
})
except Exception as e:
tool_results.append({
"type": "tool_result",
"tool_use_id": tc.id,
"content": f"Tool execution error: {str(e)}",
"is_error": True
})
else:
# Invalid tool call - report error back to Claude
error_msg = "\n".join(validation["errors"])
tool_results.append({
"type": "tool_result",
"tool_use_id": tc.id,
"content": f"Validation error: {error_msg}",
"is_error": True
})
print(f"Tool call validation failed: {error_msg}")
messages.append({"role": "user", "content": tool_results})
raise Exception(f"Exceeded max tool rounds ({max_tool_rounds}) - possible infinite loop")
Pattern 6: Loop Detection and Breaking
Claude Code can sometimes get stuck in loops — repeatedly calling the same tool with the same arguments, or producing the same response. Detecting and breaking these loops is essential for preventing wasted API calls and stuck conversations.
import hashlib
from collections import deque
class LoopDetector:
"""Detects and breaks repetitive loops in Claude Code conversations."""
def __init__(self, window_size=5, similarity_threshold=0.85):
self.window_size = window_size
self.similarity_threshold = similarity_threshold
self.action_history = deque(maxlen=window_size)
self.intervention_count = 0
self.max_interventions = 3
def record_action(self, action_type: str, action_data: dict):
"""Record an action for loop detection."""
action_hash = self._hash_action(action_type, action_data)
self.action_history.append({
"type": action_type,
"data": action_data,
"hash": action_hash
})
def detect_loop(self) -> dict:
"""Check if recent actions form a loop."""
if len(self.action_history) < 3:
return {"detected": False}
recent = list(self.action_history)
# Check for exact repetition
hashes = [a["hash"] for a in recent]
if len(set(hashes)) == 1 and len(hashes) >= 3:
return {
"detected": True,
"type": "exact_repetition",
"message": "Claude is repeating the exact same action"
}
# Check for alternating pattern (A-B-A-B)
if len(recent) >= 4:
if recent[-1]["hash"] == recent[-3]["hash"] and \
recent[-2]["hash"] == recent[-4]["hash"]:
return {
"detected": True,
"type": "alternating",
"message": "Claude is stuck in an alternating pattern"
}
# Check for high similarity in tool calls
tool_actions = [a for a in recent if a["type"] == "tool_call"]
if len(tool_actions) >= 3:
similarity = self._calculate_similarity(tool_actions)
if similarity > self.similarity_threshold:
return {
"detected": True,
"type": "similar_tools",
"message": f"Tool calls are {similarity*100:.0f}% similar",
"similarity": similarity
}
return {"detected": False}
def get_intervention_message(self, loop_info: dict) -> str:
"""Generate an intervention message to break the loop."""
self.intervention_count += 1
if self.intervention_count > self.max_interventions:
return "ESCALATE" # Signal to escalate to human or abort
interventions = [
f"I notice we may be going in circles ({loop_info['type']}). "
f"Let's try a completely different approach to solve this problem.",
f"We seem stuck in a loop. Please step back and reconsider "
f"the problem from scratch. What's a fundamentally different "
f"strategy we could try?",
f"This is the third time we've hit this loop. Please provide "
f"a summary of what you've tried so far and explain why each "
f"approach failed, then propose a new solution."
]
idx = min(self.intervention_count - 1, len(interventions) - 1)
return interventions[idx]
def _hash_action(self, action_type, action_data):
content = f"{action_type}:{json.dumps(action_data, sort_keys=True)}"
return hashlib.md5(content.encode()).hexdigest()
def _calculate_similarity(self, actions):
"""Calculate average pairwise similarity of tool call inputs."""
if len(actions) < 2:
return 0.0
similarities = []
for i in range(len(actions) - 1):
sim = self._jaccard_similarity(
actions[i]["data"],
actions[i + 1]["data"]
)
similarities.append(sim)
return sum(similarities) / len(similarities)
def _jaccard_similarity(self, dict1, dict2):
"""Calculate Jaccard similarity between two dictionaries."""
set1 = set(json.dumps(dict1, sort_keys=True).split())
set2 = set(json.dumps(dict2, sort_keys=True).split())
if not set1 and not set2:
return 1.0
intersection = set1 & set2
union = set1 | set2
return len(intersection) / len(union) if union else 0.0
# Usage in conversation loop
loop_detector = LoopDetector(window_size=6)
def conversation_with_loop_detection(messages, user_input):
messages.append({"role": "user", "content": user_input})
while True:
response = call_claude_with_retry(messages)
messages.append({"role": "assistant", "content": response.content})
# Check for tool calls
tool_uses = [b for b in response.content if b.type == "tool_use"]
if not tool_uses:
break # No tools, conversation is complete
for tu in tool_uses:
loop_detector.record_action("tool_call", {"name": tu.name, "input": tu.input})
loop_check = loop_detector.detect_loop()
if loop_check["detected"]:
intervention = loop_detector.get_intervention_message(loop_check)
if intervention == "ESCALATE":
print("Max interventions reached. Escalating to human operator.")
break
print(f"Loop detected: {loop_check['message']}")
messages.append({"role": "user", "content": intervention})
continue
# Execute tools and continue
tool_results = execute_tools(tool_uses)
messages.append({"role": "user", "content": tool_results})
return messages
Pattern 7: Dead Letter Queue for Failed Operations
In production systems, some operations will fail permanently regardless of retry strategies. Rather than losing these requests entirely, a dead letter queue (DLQ) stores failed operations for later inspection, manual retry, or debugging. This is particularly valuable for batch processing pipelines that use Claude Code.
import json
import sqlite3
from datetime import datetime
class DeadLetterQueue:
"""Stores failed Claude Code operations for later recovery."""
def __init__(self, db_path="claude_dlq.db"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS failed_operations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
operation_type TEXT NOT NULL,
payload TEXT NOT NULL,
error_message TEXT NOT NULL,
retry_count INTEGER DEFAULT 0,
status TEXT DEFAULT 'pending',
last_retry TEXT
)
""")
self.conn.commit()
def enqueue(self, operation_type: str, payload: dict, error: str):
"""Add a failed operation to the DLQ."""
self.conn.execute(
"INSERT INTO failed_operations (timestamp, operation_type, payload, error_message) "
"VALUES (?, ?, ?, ?)",
(datetime.now().isoformat(), operation_type, json.dumps(payload), str(error))
)
self.conn.commit()
print(f"Operation enqueued to DLQ: {operation_type}")
def get_pending(self, limit=10):
"""Retrieve pending operations for retry."""
cursor = self.conn.execute(
"SELECT id, operation_type, payload, error_message, retry_count "
"FROM failed_operations WHERE status = 'pending' "
"ORDER BY timestamp ASC LIMIT ?",
(limit,)
)
return cursor.fetchall()
def mark_resolved(self, op_id):
"""Mark an operation as successfully resolved."""
self.conn.execute(
"UPDATE failed_operations SET status = 'resolved' WHERE id = ?",
(op_id,)
)
self.conn.commit()
def mark_failed(self, op_id, error):
"""Mark a retry attempt as failed."""
self.conn.execute(
"UPDATE failed_operations SET retry_count = retry_count + 1, "
"last_retry = ?, error_message = ? WHERE id = ?",
(datetime.now().isoformat(), str(error), op_id)
)
self.conn.commit()
def process_queue(self, retry_func, max_retries=3):
"""Attempt to process all pending operations."""
pending = self.get_pending(limit=50)
resolved = 0
for op_id, op_type, payload_str, error, retry_count in pending:
if retry_count >= max_retries:
self.conn.execute(
"UPDATE failed_operations SET status = 'exhausted' WHERE id = ?",
(op_id,)
)
continue
payload = json.loads(payload_str)
try:
retry_func(op_type, payload)
self.mark_resolved(op_id)
resolved += 1
print(f"Resolved operation {op_id}")
except Exception as e:
self.mark_failed(op_id, e)
print(f"Retry failed for operation {op_id}: {e}")
self.conn.commit()
print(f"Processed {len(pending)} operations, resolved {resolved}")
return resolved
# Usage
dlq = DeadLetterQueue()
def process_with_dlq(operation_type, payload, claude_func):
"""Wrap a Claude operation with DLQ fallback."""
try:
return claude_func(payload)
except Exception as e:
# All retries exhausted, send to DLQ
dlq.enqueue(operation_type, payload, e)
raise
# Later, run a recovery job
def retry_operation(op_type, payload):
if op_type == "code_review":
return call_claude_with_retry(payload["messages"])
elif op_type == "doc_generation":
return call_claude_with_retry(payload["messages"])
# Schedule this periodically
dlq.process_queue(retry_operation, max_retries=3)
Pattern 8: Graceful Degradation
When all recovery attempts fail, graceful degradation ensures your application still provides value. Instead of showing an error, you fall back to cached results, simplified responses, or pre-computed answers. The user experience degrades but doesn't break.
class GracefulDegradation:
"""Provides degraded but functional responses when Claude is unavailable."""
def __init__(self):
self.response_cache = {}
self.static_fallbacks = {
"code_review": "I'm unable to perform a detailed code review right now. "
"Here are general best practices to check: ensure proper "
"error handling, validate inputs, and add tests for edge cases.",
"doc_generation": "Documentation generation is temporarily unavailable. "
"Please refer to the inline comments and function signatures "
"for documentation in the meantime.",
"general": "I'm experiencing technical difficulties. Please try again in a moment."
}
def get_response(self, request_type, prompt, context=None):
"""Try full response, then degrade gracefully."""
# Level 1: Try the full Claude response
try:
result = call_claude_with_retry(
self._build_messages(prompt, context),
max_retries=3
)
response_text = result.content[0].text
# Cache successful responses
cache_key = self._cache_key(request_type, prompt)
self.response_cache[cache_key] = {
"response": response_text,
"timestamp": time.time(),
"quality": "full"
}
return {"text": response_text, "quality": "full", "source": "claude"}
except Exception as e:
print(f"Full response failed: {e}")
# Level 2: Try a cached response
cache_key = self._cache_key(request_type, prompt)
if cache_key in self.response_cache:
cached = self.response_cache[cache_key]
print("Serving cached response")
return {
"text": cached["response"],
"quality": "cached",
"source": "cache",
"cached_at": cached["timestamp"]
}
# Level 3: Try a simplified response with smaller model
try:
result = client.messages.create(
model="claude-3-5-haiku-20241022",
max_tokens=512,
messages=[{"role": "user", "content": f"Briefly: {prompt}"}]
)
return {
"text": result.content[0].text,
"quality": "degraded",
"source": "claude_haiku"
}
except Exception:
pass
# Level 4: Static fallback
fallback = self.static_fallbacks.get(request_type, self.static_fallbacks["general"])
return {
"text": fallback,
"quality": "static_fallback",
"source": "fallback"
}
def _build_messages(self, prompt, context):
messages = [{"role": "user", "content": prompt}]
if context and context.get("history"):
messages = context["history"] + messages
return messages
def _cache_key(self, request_type, prompt):
return f"{request_type}:{hashlib.md5(prompt.encode()).hexdigest()}"
Combining Patterns: A Resilient Claude Code Wrapper
In practice, you'll want to combine multiple patterns into a single resilient wrapper. Here's how to integrate retry, circuit breaker, loop detection, context recovery, and graceful degradation into one cohesive system.
class ResilientClaudeCode:
"""
Production-ready wrapper combining all error recovery patterns.
"""
def __init__(self, config=None):
self.config = config or {}
self.client = Anthropic()
self.circuit_breaker = CircuitBreaker(
failure_threshold=self.config.get("failure_threshold", 5),
recovery_timeout=self.config.get("recovery_timeout", 60)
)
self.context_manager = ContextWindowManager(
max_tokens=self.config.get("max_context_tokens", 180000)
)
self.loop_detector = LoopDetector(
window_size=self.config.get("loop_window", 6)
)
self.degradation = GracefulDegradation()
self.dlq = DeadLetterQueue()
def chat(self, messages, user_input, request_type="general"):
"""Process a chat turn with full error recovery."""
# Add user input
messages.append({"role": "user", "content": user_input})
# Recover context if needed
try:
messages = self.context_manager.recover_context(messages)
except Exception as e:
print(f"Context recovery failed, continuing with truncated context: {e}")
messages = messages[-10:] # Keep last 10 messages as fallback
# Attempt response with circuit breaker
try:
response = self.circuit_breaker.call(self._make_request, messages)
assistant_content = response.content
messages.append({"role": "assistant", "content": assistant_content})
# Handle tool use with loop detection
if any(b.type == "tool_use" for b in assistant_content):
messages = self._handle_tool_use(messages, response)
return {
"messages": messages,
"response": response,
"quality": "full"
}
except Exception as e:
# All recovery failed - enqueue to DLQ and degrade
print(f"Primary request failed: {e}")
self.dlq.enqueue(request_type, {
"messages": messages,
"user_input": user_input
}, str(e))
# Graceful degradation
degraded = self.degradation.get_response(
request_type, user_input
)
messages.append({"role": "assistant", "content": degraded["text"]})
return {
"messages": messages,
"response": degraded,
"quality": degraded["quality"]
}
def _make_request(self, messages):
"""Make a Claude API request with retry logic."""
return call_claude_with_retry(
messages,
max_retries=self.config.get("max_retries", 3)
)
def _handle_tool_use(self, messages, response):
"""Handle tool use with validation and loop detection."""
max_rounds = self.config.get("max_tool_rounds", 10)
recovery = ToolCallRecovery(
self.config.get("tools", [])
)
for _ in range(max_rounds):
tool_uses = [b for b in response.content if b.type == "tool_use"]
if not tool_uses:
break
# Record actions for loop detection
for tu in tool_uses:
self.loop_detector.record_action(
"tool_call",
{"name": tu.name, "input": tu.input}
)
# Check for loops
loop = self.loop_detector.detect_loop()
if loop["detected"]:
intervention = self.loop_detector.get_intervention_message(loop)
if intervention == "ESCALATE":
print("Loop escalation - breaking tool use cycle")
break
messages.append({"role": "user", "content": intervention})
else:
# Execute tools
tool_results = []
for tu in tool_uses:
validation = recovery.validate_tool_call(tu.name, tu.input)
if validation["valid"]:
try:
result = execute_tool(tu.name, tu.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": tu.id,
"content": str(result)
})
except Exception as e:
tool_results.append({
"type": "tool_result",
"tool_use_id": tu.id,
"content": f"Error: {e}",
"is_error": True
})
else:
tool_results.append({
"type": "tool_result",
"tool_use_id": tu.id,