When you first integrate the Claude API, it feels complete the moment anthropic.Anthropic().messages.create(...) returns a response. Once you put that call inside a real service, though, a much larger world opens up: sudden 429s, flaky connections, scattered 503s, hanging requests. How you handle these determines the quality of your production service.
This article is the design I settled on after running the Claude API inside a Python backend for a while. I am skipping the basics and going straight to the reliability layer most guides omit.
Retry design breaks first in production, timeout design breaks second
When services that use the Claude API have incidents, the cause is almost always one of three things:
- Sloppy retries that cannot absorb transient errors
- Coarse timeouts that let one-way hangs drag the whole service down
- Misreading rate-limit responses, which turns a spike into a cascade
Once you have these three in focus, wrapping the SDK is not enough. The SDK is built to handle the happy path; you have to build the unhappy path yourself.
Decide what retries target before writing any retry code
The single most important decision in retry design is defining what you retry. Errors from the Claude API fall into these buckets:
- Retry: 429 (rate limit), 503 (temporary overload), connection errors, read timeouts
- Do not retry: most 4xx (malformed request), 401 (auth), 403 (permission), context overflow
- Case by case: 500 (server), 504 (gateway timeout)
Ignoring this split and writing "retry three times on any error" creates the classic incident where a brief auth outage triples your error load simply from the retry attempts.
My baseline looks like this:
import random
import time
from anthropic import Anthropic, APIStatusError, APIConnectionError, APITimeoutError
client = Anthropic(timeout=60.0, max_retries=0)
RETRYABLE_STATUS = {429, 503, 504, 529}
MAX_ATTEMPTS = 5
BASE_DELAY = 1.0
MAX_DELAY = 30.0
def call_with_retry(payload: dict):
attempt = 0
while True:
attempt += 1
try:
return client.messages.create(**payload)
except APIStatusError as e:
if e.status_code not in RETRYABLE_STATUS or attempt >= MAX_ATTEMPTS:
raise
sleep_for = _retry_delay(e, attempt)
except (APIConnectionError, APITimeoutError):
if attempt >= MAX_ATTEMPTS:
raise
sleep_for = _backoff(attempt)
time.sleep(sleep_for)
def _backoff(attempt: int) -> float:
cap = min(BASE_DELAY * 2 ** (attempt - 1), MAX_DELAY)
return random.uniform(0, cap)
def _retry_delay(err: APIStatusError, attempt: int) -> float:
retry_after = getattr(err.response, "headers", {}).get("retry-after")
if retry_after:
try:
return min(float(retry_after), MAX_DELAY)
except ValueError:
pass
return _backoff(attempt)I intentionally set the SDK's max_retries to 0 and handle retries myself for two reasons. First, I want to read and log the Retry-After header. Second, I want every sleep to happen in one place so I can reason about timing during incidents. When retries are hidden inside the SDK you lose visibility into where your seconds are going.
When you see a 429, find out who emitted it and why
A 429 is Anthropic telling you to slow down, but the "who" and "why" behind it vary. Account-wide token rate, per-model RPM, concurrent connection caps on a single key — these are different signals surfaced the same way.
What saves you later is logging status errors with structure, including the Retry-After header and the request ID:
except APIStatusError as e:
log.warning({
"event": "anthropic_status_error",
"status": e.status_code,
"message": getattr(e, "message", None),
"request_id": getattr(e, "request_id", None),
"retry_after": getattr(e.response, "headers", {}).get("retry-after"),
"attempt": attempt,
})The request_id is decisive if you ever contact Anthropic support. Teams that forget to log it lose the ability to correlate server-side traces with their incidents.
In one incident I had to debug, sporadic late-night 429s were appearing. Because I had request IDs and timestamps, I could see that the spikes were confined to one specific model. My application had not changed, so I could conclude quickly that the issue was not on our side.
Separate connect timeout from read timeout
Python's httpx lets you configure connect and read timeouts independently. For a service like the Claude API, where request time varies wildly with prompt size and model, this separation is not optional.
Connect timeout is about "can we establish a TCP connection at all," which should fail fast. I use 5 seconds. Read timeout covers "Claude is thinking," which should be generous. Depending on model and max_tokens, I set 60 to 180 seconds.
import httpx
from anthropic import Anthropic
http_client = httpx.Client(
timeout=httpx.Timeout(
connect=5.0,
read=120.0,
write=10.0,
pool=5.0,
),
limits=httpx.Limits(max_connections=50, max_keepalive_connections=20),
)
client = Anthropic(http_client=http_client, max_retries=0)Another subtle point: passing a scalar timeout=60.0 applies to every phase including connect. A 60-second connect timeout is too long — it masks DNS and TCP issues. Always split it.
Streaming adds another wrinkle. With streaming you have two concepts: time to first byte and time to completion. Read timeout is effectively the idle interval between chunks, so 60 seconds is usually fine. For non-streaming long outputs, 120 to 180 is safer.
Design retry and timeout together
This is where people trip. Setting read timeout to 120 and retries to 5 means the worst case is over ten minutes. Most upstream requests have SLAs of seconds to tens of seconds. Exceeding that inside your API layer creates incidents.
My rules:
- Cap total time at 70% of the upstream SLA
- Work out per-attempt timeout × max retries so they stay under the total cap
- Limit user-facing synchronous calls to
max_attempts=3— 5+ belongs to async queues
Concretely:
import time
def call_with_budget(payload: dict, total_budget_s: float):
start = time.monotonic()
attempt = 0
while True:
attempt += 1
remaining = total_budget_s - (time.monotonic() - start)
if remaining <= 0:
raise TimeoutError("budget exhausted")
per_attempt_timeout = min(remaining, 30.0)
try:
return client.with_options(timeout=per_attempt_timeout).messages.create(**payload)
except (APIConnectionError, APITimeoutError):
if attempt >= MAX_ATTEMPTS or (time.monotonic() - start) >= total_budget_s:
raise
time.sleep(_backoff(attempt))with_options(timeout=...) lets you shrink the per-attempt budget as remaining time shrinks. With this pattern you never blow the upstream SLA.
Concurrency — do not burn your rate limit silently
Batch jobs and multi-tenant services that fan out calls always hit 429 if concurrency is uncapped. asyncio.gather on 50 requests is a production anti-pattern.
I use asyncio.Semaphore for an explicit parallelism ceiling and layer a token bucket on top for per-second rate control.
import asyncio
from anthropic import AsyncAnthropic
client = AsyncAnthropic()
sem = asyncio.Semaphore(10)
async def call_async(payload: dict):
async with sem:
return await client.messages.create(**payload)A semaphore alone will not cap requests per second. When bursts are expected, add a bucket:
class TokenBucket:
def __init__(self, rate: float, capacity: int):
self.rate = rate
self.capacity = capacity
self.tokens = capacity
self.last = time.monotonic()
self.lock = asyncio.Lock()
async def acquire(self):
async with self.lock:
now = time.monotonic()
elapsed = now - self.last
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last = now
if self.tokens < 1:
wait = (1 - self.tokens) / self.rate
await asyncio.sleep(wait)
self.tokens = 0
else:
self.tokens -= 1I target 70 to 80 percent of the Anthropic RPM limit. That headroom protects against other clients sharing the same key and against small demand spikes.
Standardize your error logging
The single biggest operational difference between mature and immature services is the granularity of error logs. Do not re-raise SDK exceptions bare — attach metadata and write structured logs first.
def call_safely(payload: dict):
try:
return client.messages.create(**payload)
except APIStatusError as e:
log.error({
"event": "claude_api_status_error",
"status": e.status_code,
"request_id": getattr(e, "request_id", None),
"payload_model": payload.get("model"),
"payload_max_tokens": payload.get("max_tokens"),
"input_tokens_estimate": _estimate_tokens(payload),
})
raise
except (APIConnectionError, APITimeoutError) as e:
log.error({
"event": "claude_api_network_error",
"error_type": type(e).__name__,
})
raiseRequest ID, model, max_tokens, and an estimated input token count — those four fields let you answer "which request was heavy" and "which model was unstable" months later. Without them, the data is gone.
Three production failure modes worth knowing
Mode 1: silent streaming disconnects. When the network drops mid-stream, some clients simply stop emitting events rather than raising an error. Add an idle guard:
async def stream_with_idle_guard(payload, idle_limit_s=20.0):
last = time.monotonic()
async with client.messages.stream(**payload) as stream:
async for event in stream:
now = time.monotonic()
if now - last > idle_limit_s:
raise TimeoutError("stream idle exceeded")
last = now
yield eventMode 2: context overflow from user-generated input. Services that feed raw user content into Claude eventually start getting context overflow errors. Check input size before the API call and summarize or chunk when it gets close.
Mode 3: retry storms from ignored Retry-After. When a 429 comes with a Retry-After value and your code keeps its own exponential backoff without reading the header, you fight the server's guidance. Respect Retry-After first, backoff second.
Closing thought
Running the Claude API in production is about quiet defenses, not clever optimizations. Split connect and read timeouts. Honor Retry-After. Log request IDs. Work back from an overall time budget. These simple habits move the reliability needle more than any tuning trick.
If you only add one thing today, add request-ID logging. The next time something goes wrong, you will be glad you did.