Your Claude wrapper works for the first month. It works through your first thousand users. Then Tuesday afternoon happens — traffic doubles for no obvious reason, your primary path hits its rate limit, your naive retry loop stampedes Anthropic's recovery, and the site is down for eleven minutes while you frantically Google "how to handle 429 errors." This post is what you wish you had built the first time.
The wrapper below is what teams that have been through the outage converge on. It composes four independent reliability primitives — a client-side rate limiter, a jittered retry loop, a circuit breaker, and multi-provider fallback — into one class you can drop into any Python service. Every technique is drawn from patterns tested in production. The complete implementation is ~250 lines. You can copy it, adapt it, ship it.
The five failure modes a wrapper must handle
Before the code, the shape of the problem. Any wrapper claiming to be "production-ready" must handle every one of these:
- Rate limits — 429 responses when you exceed RPM or TPM. Guaranteed to happen at scale. Solved by client-side smoothing (proactive) and jittered retry (reactive).
- Overload errors — 529 on Anthropic when the provider itself is under load. Not your fault, but you still have to handle it.
- Sustained degradation — the provider is having a bad hour. Retrying every request is throwing gasoline on the fire. Solved by circuit breakers.
- Full provider outages — Anthropic direct is down. You need to route to Bedrock or a different model without user-visible failures.
- Your own bugs — 400s from malformed requests. Retrying makes them worse. The wrapper must distinguish these from transient failures.
A wrapper that handles only rate limits is fragile. A wrapper that handles all five is the thing you actually want. Here's how each one is built.
Requirements checklist
Before building, be explicit about what "bulletproof" means. The wrapper below satisfies every item:
- Never retries non-retryable errors (400, 401, 403, 404, 422)
- Always honors Retry-After when provided by the server
- Uses full-jitter exponential backoff to prevent stampedes
- Caps individual retry delay at 60 seconds and total attempts at 5
- Applies client-side rate limiting to avoid tripping 429s proactively
- Opens a circuit breaker after sustained failure to save resources
- Falls back to Bedrock when Anthropic direct is degraded
- Emits metrics on every retry, fallback, and breaker transition
- Fails cleanly — never hangs the caller indefinitely
- Distinguishes provider errors from your own bugs
Layer 1: Error classification
Everything starts here. The single most common bug in AI wrappers is retrying errors that will never succeed. Retrying a 400 wastes tokens and time; retrying a 401 wastes support tickets. The wrapper's first responsibility is knowing what to retry and what to surface immediately.
from anthropic import (
RateLimitError, APIStatusError, APIConnectionError, BadRequestError
)
def is_retryable(exception) -> bool:
# Never retry the caller's fault
if isinstance(exception, BadRequestError):
return False # 400 — malformed request
if isinstance(exception, APIStatusError):
code = exception.status_code
if code in (401, 403, 404, 422):
return False # auth/permission/not-found/validation
if code in (429, 500, 502, 503, 504, 529):
return True # rate limit or transient server error
if isinstance(exception, RateLimitError):
return True
if isinstance(exception, APIConnectionError):
return True # network — retry
return False # unknown — don't retry unknowns
def is_breaker_signal(exception) -> bool:
# A subset of retryable errors that signal provider degradation
if isinstance(exception, RateLimitError):
return True
if isinstance(exception, APIStatusError):
return exception.status_code in (500, 502, 503, 504, 529)
if isinstance(exception, (APIConnectionError, ConnectionError, TimeoutError)):
return True
return FalseNote the distinction between is_retryable and is_breaker_signal. Rate limits are retryable but they're your fault (you sent too much traffic), so they shouldn't count against the provider's health score. Server errors are both retryable and signal the provider is struggling — those count.
Layer 2: Client-side rate limiting
The point of client-side rate limiting is not to enforce a limit you don't have — it's to smooth your traffic so you never hit the ceiling. A token bucket refills at a fixed rate and every request costs a proportional amount from the bucket. Requests that exceed available tokens wait.
import time, threading
class TokenBucket:
def __init__(self, tpm: int, burst_multiplier: float = 1.5):
self.rate_per_sec = tpm / 60.0
self.capacity = int(tpm * burst_multiplier / 60.0 * 60.0)
self.tokens = float(self.capacity)
self.last_refill = time.time()
self.lock = threading.Lock()
def acquire(self, cost: int):
with self.lock:
self._refill()
if self.tokens >= cost:
self.tokens -= cost
return
# Not enough — sleep until refilled
deficit = cost - self.tokens
wait = deficit / self.rate_per_sec
time.sleep(wait)
with self.lock:
self._refill()
self.tokens -= cost
def _refill(self):
now = time.time()
elapsed = now - self.last_refill
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate_per_sec)
self.last_refill = nowSize the bucket at 80-85% of your tier's actual limit. The buffer accounts for measurement imprecision and gives the provider room for its own smoothing. Setting the bucket at 100% causes constant near-miss 429s.
Layer 3: Full-jitter exponential backoff
When a retry is warranted, backoff prevents the stampede. If a hundred clients all retry after exactly 2 seconds, they all retry at the same moment, and the provider's recovery is immediately re-broken.
import random
def full_jitter_delay(attempt: int, base: float = 1.0, cap: float = 60.0) -> float:
# Every retry lands somewhere in [0, exp].
# Spreads coordinated retries evenly across the backoff window.
exp = min(cap, base * (2 ** attempt))
return random.uniform(0, exp)
def parse_retry_after(exception) -> float | None:
response = getattr(exception, "response", None)
if not response:
return None
value = response.headers.get("retry-after")
if not value:
return None
try:
return float(value)
except ValueError:
return NoneWhen the server provides Retry-After, honor it — it's precise information from the provider about when the limit will clear. Cap the value even so; a misconfigured proxy occasionally sends Retry-After: 86400.
Layer 4: The circuit breaker
Retries protect individual requests. The circuit breaker protects your service. When failures pile up in a window, all subsequent requests fail fast without touching the provider. This saves tokens, latency, and lets the provider recover.
from enum import Enum
from collections import deque
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class CircuitOpenError(Exception):
pass
class CircuitBreaker:
def __init__(self, name: str, threshold: float = 0.5,
window_seconds: float = 30.0, min_samples: int = 20,
reset_timeout: float = 30.0):
self.name = name
self.threshold = threshold
self.window = window_seconds
self.min_samples = min_samples
self.reset_timeout = reset_timeout
self.state = CircuitState.CLOSED
self.events: deque = deque()
self.reopens_at: float | None = None
self.consecutive_failures = 0
self.lock = threading.Lock()
def allow_request(self) -> bool:
with self.lock:
now = time.time()
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if self.reopens_at and now >= self.reopens_at:
self.state = CircuitState.HALF_OPEN
return True
return False
if self.state == CircuitState.HALF_OPEN:
return True # probing
def record_success(self):
with self.lock:
self.events.append((time.time(), True))
self._prune()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
self.consecutive_failures = 0
def record_failure(self):
with self.lock:
now = time.time()
self.events.append((now, False))
self._prune()
if self.state == CircuitState.HALF_OPEN:
# Probe failed — back to open with longer timeout
self.consecutive_failures += 1
self.state = CircuitState.OPEN
self.reopens_at = now + min(
300.0, self.reset_timeout * (2 ** self.consecutive_failures)
)
return
if self.state == CircuitState.CLOSED:
total = len(self.events)
failures = sum(1 for _, ok in self.events if not ok)
if total >= self.min_samples and failures / total >= self.threshold:
self.state = CircuitState.OPEN
self.reopens_at = now + self.reset_timeout
def _prune(self):
cutoff = time.time() - self.window
while self.events and self.events[0][0] < cutoff:
self.events.popleft()Three states, one lock, sliding window. Trips at 50% failure rate over 30 seconds with a minimum of 20 samples (prevents false positives in quiet periods). Reset timeout backs off exponentially on repeated open transitions, capped at 5 minutes.
Layer 5: Multi-provider fallback
When the primary path is exhausted — either the circuit is open or retries are done — the wrapper falls back to Bedrock. Same model, different provider, independent quota pool. Usually works when Anthropic direct is having a bad day.
import boto3, json as jsonlib
from botocore.config import Config
def make_bedrock_client():
config = Config(retries={"max_attempts": 3, "mode": "adaptive"})
return boto3.client("bedrock-runtime", region_name="us-east-1", config=config)
def call_bedrock_fallback(messages: list, model: str = "anthropic.claude-sonnet-4-6-v1:0",
max_tokens: int = 1024) -> dict:
client = make_bedrock_client()
response = client.invoke_model(
modelId=model,
body=jsonlib.dumps({
"messages": messages,
"max_tokens": max_tokens,
"anthropic_version": "bedrock-2023-05-31",
}),
)
return jsonlib.loads(response["body"].read())Putting it together: the BulletproofClaude class
Every piece from above, composed into one class. This is what you drop into your codebase.
import logging
from anthropic import Anthropic
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class WrapperConfig:
tpm: int = 40_000 # Tier-1 default; raise for higher tiers
max_retries: int = 5
base_backoff: float = 1.0
cap_backoff: float = 60.0
retry_after_cap: float = 300.0
breaker_threshold: float = 0.5
breaker_window: float = 30.0
breaker_min_samples: int = 20
breaker_reset_timeout: float = 30.0
class BulletproofClaude:
def __init__(self, config: WrapperConfig | None = None):
self.config = config or WrapperConfig()
self.client = Anthropic()
self.bucket = TokenBucket(tpm=int(self.config.tpm * 0.85))
self.breaker = CircuitBreaker(
name="anthropic-direct",
threshold=self.config.breaker_threshold,
window_seconds=self.config.breaker_window,
min_samples=self.config.breaker_min_samples,
reset_timeout=self.config.breaker_reset_timeout,
)
def call(self, messages: list, model: str = "claude-sonnet-4-6",
max_tokens: int = 1024, correlation_id: str = "") -> dict:
# Step 1: pre-flight budget check
estimated = self._estimate_tokens(messages, max_tokens)
self.bucket.acquire(estimated)
# Step 2: circuit breaker check
if not self.breaker.allow_request():
logger.info("breaker_open", extra={"correlation_id": correlation_id})
return self._fallback(messages, model, max_tokens, correlation_id)
# Step 3: retry loop
last_exception: Exception | None = None
for attempt in range(self.config.max_retries):
try:
response = self.client.messages.create(
model=model,
max_tokens=max_tokens,
messages=messages,
)
self.breaker.record_success()
return {
"content": [b.model_dump() for b in response.content],
"provider": "anthropic",
"model": model,
"usage": {
"input_tokens": response.usage.input_tokens,
"output_tokens": response.usage.output_tokens,
},
}
except Exception as e:
last_exception = e
if not is_retryable(e):
if is_breaker_signal(e):
self.breaker.record_failure()
raise
if is_breaker_signal(e):
self.breaker.record_failure()
if attempt == self.config.max_retries - 1:
break # exhausted
retry_after = parse_retry_after(e)
if retry_after is not None:
wait = min(retry_after, self.config.retry_after_cap)
else:
wait = full_jitter_delay(attempt, self.config.base_backoff,
self.config.cap_backoff)
logger.info("retry", extra={
"correlation_id": correlation_id,
"attempt": attempt + 1, "wait_sec": wait,
"error": type(e).__name__,
})
time.sleep(wait)
# Step 4: retries exhausted — fall back
logger.warning("retries_exhausted", extra={
"correlation_id": correlation_id,
"last_error": type(last_exception).__name__ if last_exception else None,
})
return self._fallback(messages, model, max_tokens, correlation_id)
def _fallback(self, messages, model, max_tokens, correlation_id):
try:
bedrock_model = f"anthropic.{model}-v1:0" if "claude" in model else model
result = call_bedrock_fallback(messages, bedrock_model, max_tokens)
logger.info("fallback_success", extra={
"correlation_id": correlation_id,
"provider": "bedrock",
})
return {
"content": result.get("content", []),
"provider": "bedrock",
"model": bedrock_model,
"usage": result.get("usage", {}),
}
except Exception as e:
logger.error("fallback_failed", extra={
"correlation_id": correlation_id,
"error": type(e).__name__,
})
raise
def _estimate_tokens(self, messages, max_output_tokens):
try:
r = self.client.messages.count_tokens(
model="claude-sonnet-4-6", messages=messages
)
return r.input_tokens + max_output_tokens
except Exception:
# Fall back to rough estimate if count_tokens fails
chars = sum(len(str(m.get("content", ""))) for m in messages)
return int(chars / 4) + max_output_tokensUsing it
import uuid
from bulletproof_claude import BulletproofClaude, WrapperConfig
# Configure once per service
wrapper = BulletproofClaude(WrapperConfig(
tpm=80_000, # Tier 2 budget
max_retries=5,
breaker_threshold=0.5,
))
# Call from wherever you'd normally call Anthropic
def summarize(text: str) -> str:
result = wrapper.call(
messages=[{"role": "user", "content": f"Summarize:\n\n{text}"}],
model="claude-sonnet-4-6",
max_tokens=500,
correlation_id=str(uuid.uuid4()),
)
# content is the standard Anthropic format
return result["content"][0]["text"]What this wrapper actually gives you
- Never hits 429 unnecessarily — the token bucket smooths outgoing traffic below your tier limit.
- Handles transient failures gracefully — jittered retries with respect for
Retry-After. - Protects against outages — the breaker opens after sustained failure, saving tokens during recovery.
- Fails over to Bedrock automatically — either when the breaker is open or when direct retries are exhausted.
- Never retries your bugs — 400s and 401s surface immediately for developer attention.
- Never hangs the caller — every path either returns a response or raises within bounded time.
- Emits actionable logs — correlation IDs and structured fields let you trace any request end-to-end.
What it doesn't give you (yet)
This is the minimum viable version. For higher-scale services, add these on top:
- Distributed state — the token bucket and breaker are per-process. For multi-instance services, back them with Redis so all instances share limits and health signals.
- Metrics emission — wire the log points to Prometheus or Datadog. Track retry rate, breaker state transitions, fallback usage, and cache hit rate.
- Prompt caching — add
cache_controlmarkers to reduce cost for stable prefixes. - Cross-provider fallback — the current wrapper falls back to Bedrock (same model). For catastrophic outages, add GPT or Gemini as a tertiary tier. See the fallback strategies deep-dive linked below.
- Context management — long conversations need history compression before hitting the wrapper. See the context window guide.
Where to go deeper
Each layer in this wrapper is a topic worth understanding on its own. If you're building at scale, the following pillar guides cover each one in detail with production considerations this post skims:
- Handling Rate Limits Across AI APIs — every provider's tier structure, TPM/RPM math, batch API strategies, tier progression rules.
- Retry Logic Patterns That Actually Work — the deeper math of backoff strategies, idempotency, retry budgets, distributed coordination.
- Circuit Breakers for AI APIs — state machine details, half-open probe patterns, multi-tier breakers, Redis-backed distributed variants.
- Fallback Strategies: Claude → GPT → Gemini — provider-neutral prompts, tool portability, quality equivalence measurement, gateway options.