Rate limits are the most common reason production LLM workloads fail silently or degrade unexpectedly. If you need to avoid Claude rate limits consistently, you need a layered approach: understand the exact limits, implement proper backoff, control concurrency, monitor headers, and have a fallback path. This guide walks through each layer with runnable code you can drop into a Python service today.
Step 1: Understand the exact limits you’re working against
Anthropic enforces limits at three levels: requests per minute (RPM), tokens per minute (TPM), and tokens per day (TPD). The limits vary by model and your organization’s tier. As of writing, the public tiers look roughly like this:
| Tier | Claude 3.5 Sonnet RPM | Claude 3.5 Sonnet TPM | Claude 3 Opus RPM | Claude 3 Opus TPM |
|---|---|---|---|---|
| Build | 50 | 40,000 | 20 | 10,000 |
| Scale | 1,000 | 800,000 | 200 | 200,000 |
These numbers change. The authoritative source is the response headers on every request. Always read anthropic-ratelimit-requests-limit, anthropic-ratelimit-requests-remaining, anthropic-ratelimit-tokens-limit, and anthropic-ratelimit-tokens-remaining. Treat the headers as ground truth, not the documentation.
import httpx
from dataclasses import dataclass
@dataclass
class RateLimitSnapshot:
requests_limit: int
requests_remaining: int
tokens_limit: int
tokens_remaining: int
reset_requests_at: str # RFC 3339 timestamp
reset_tokens_at: str
def parse_rate_limit_headers(headers: httpx.Headers) -> RateLimitSnapshot:
return RateLimitSnapshot(
requests_limit=int(headers.get("anthropic-ratelimit-requests-limit", 0)),
requests_remaining=int(headers.get("anthropic-ratelimit-requests-remaining", 0)),
tokens_limit=int(headers.get("anthropic-ratelimit-tokens-limit", 0)),
tokens_remaining=int(headers.get("anthropic-ratelimit-tokens-remaining", 0)),
reset_requests_at=headers.get("anthropic-ratelimit-requests-reset", ""),
reset_tokens_at=headers.get("anthropic-ratelimit-tokens-reset", ""),
)Verify: Make a single request to messages.create and log the parsed snapshot. Confirm the numbers match your expected tier.
Step 2: Implement exponential backoff with jitter and header-aware sleep
The naive approach — sleep 60 seconds on 429 — wastes capacity and still hammers the limit when the window resets. Instead, calculate the exact wait time from the reset headers, add jitter, and back off exponentially on repeated failures.
import asyncio
import random
import time
from typing import Optional
import httpx
class AnthropicRateLimiter:
def __init__(self, client: httpx.AsyncClient, max_retries: int = 5):
self.client = client
self.max_retries = max_retries
self._request_semaphore: Optional[asyncio.Semaphore] = None
self._token_semaphore: Optional[asyncio.Semaphore] = None
async def _wait_until_reset(self, reset_header: str, buffer_seconds: float = 1.0) -> float:
"""Parse RFC 3339 reset timestamp and return seconds to wait."""
if not reset_header:
return 60.0 # fallback
try:
reset_time = time.time()
# Header format: "2024-01-15T10:30:00Z"
from datetime import datetime, timezone
dt = datetime.fromisoformat(reset_header.replace("Z", "+00:00"))
reset_time = dt.timestamp()
wait = max(0, reset_time - time.time()) + buffer_seconds
return wait
except Exception:
return 60.0
async def _exponential_backoff(self, attempt: int, base: float = 1.0, cap: float = 60.0) -> float:
"""Exponential backoff with full jitter."""
wait = min(base * (2 ** attempt), cap)
return random.uniform(0, wait)
async def request_with_backoff(self, **kwargs) -> httpx.Response:
last_exception = None
for attempt in range(self.max_retries + 1):
try:
response = await self.client.post(
"https://api.anthropic.com/v1/messages",
**kwargs
)
snapshot = parse_rate_limit_headers(response.headers)
# Success — return response even if it's a 429 (we'll handle below)
if response.status_code != 429:
return response
# 429: calculate precise wait from headers
wait_requests = await self._wait_until_reset(snapshot.reset_requests_at)
wait_tokens = await self._wait_until_reset(snapshot.reset_tokens_at)
wait = max(wait_requests, wait_tokens)
# Add jittered exponential backoff on top
backoff = await self._exponential_backoff(attempt)
total_wait = wait + backoff
print(f"Rate limited (attempt {attempt + 1}). Waiting {total_wait:.1f}s "
f"(header wait: {wait:.1f}s, backoff: {backoff:.1f}s)")
await asyncio.sleep(total_wait)
continue
except httpx.RequestError as e:
last_exception = e
backoff = await self._exponential_backoff(attempt)
print(f"Request error (attempt {attempt + 1}): {e}. Backing off {backoff:.1f}s")
await asyncio.sleep(backoff)
raise last_exception or RuntimeError("Max retries exceeded")Verify: Simulate a 429 by temporarily setting your API key to an invalid one that returns 429, or use a mock server. Confirm the client waits approximately until the reset timestamp plus jitter, not a fixed 60 seconds.
Step 3: Control concurrency with token-aware semaphores
RPM limits are easier to reason about than TPM limits because token consumption varies per request. A single 100k-token request can consume your entire minute’s token budget. Use two semaphores: one for request slots, one for token budget. Estimate tokens before sending (roughly 4 chars per token for English) and acquire token permits proportionally.
import tiktoken
class TokenAwareLimiter(AnthropicRateLimiter):
def __init__(self, client: httpx.AsyncClient, max_concurrent_requests: int = 10,
max_tokens_per_minute: int = 40000, max_retries: int = 5):
super().__init__(client, max_retries)
self._request_semaphore = asyncio.Semaphore(max_concurrent_requests)
self._token_budget = max_tokens_per_minute
self._token_semaphore = asyncio.Semaphore(max_tokens_per_minute)
self._token_refill_task: Optional[asyncio.Task] = None
self._encoding = tiktoken.get_encoding("cl100k_base") # good approximation for Claude
def estimate_tokens(self, messages: list[dict], system: str = "",
max_tokens: int = 4096) -> int:
"""Rough token estimate for request budgeting."""
text = system + "".join(m.get("content", "") for m in messages)
input_tokens = len(self._encoding.encode(text))
return input_tokens + max_tokens # assume full output budget
async def _refill_token_budget(self):
"""Refill token semaphore every 60 seconds based on current limit."""
while True:
await asyncio.sleep(60)
# In practice, read the current limit from the last response headers
# For now, refill to configured max
current = self._token_semaphore._value
if current < self._token_budget:
# Release permits up to budget
for _ in range(self._token_budget - current):
self._token_semaphore.release()
async def request_with_backoff(self, messages: list[dict], system: str = "",
max_tokens: int = 4096, **kwargs) -> httpx.Response:
estimated = self.estimate_tokens(messages, system, max_tokens)
# Acquire request slot
async with self._request_semaphore:
# Acquire token budget (blocking until available)
acquired = 0
while acquired < estimated:
try:
await asyncio.wait_for(self._token_semaphore.acquire(), timeout=0.1)
acquired += 1
except asyncio.TimeoutError:
# Check if we should give up
if self._token_semaphore._value >= estimated - acquired:
# Enough permits now available, grab them
for _ in range(estimated - acquired):
self._token_semaphore.acquire()
acquired = estimated
continue
try:
response = await super().request_with_backoff(
json={
"model": kwargs.get("model", "claude-3-5-sonnet-20241022"),
"messages": messages,
"system": system,
"max_tokens": max_tokens,
**{k: v for k, v in kwargs.items() if k not in ("model", "messages", "system", "max_tokens")}
},
headers={
"x-api-key": kwargs.get("api_key"),
"anthropic-version": "2023-06-01",
"content-type": "application/json",
}
)
# Update token budget from actual usage
snapshot = parse_rate_limit_headers(response.headers)
if snapshot.tokens_remaining >= 0:
self._token_budget = snapshot.tokens_limit
# Adjust semaphore to match actual remaining
# (simplified — in production, track a rolling window)
return response
finally:
# Return unused token permits (actual usage may be less than estimate)
# In practice, parse usage from response and return the difference
passVerify: Run a load test with 50 concurrent requests against a test endpoint. Confirm the concurrency never exceeds max_concurrent_requests and token consumption stays within the TPM limit by checking the anthropic-ratelimit-tokens-remaining header on responses.
Step 4: Implement a durable client-side queue with persistence
For workloads that can tolerate latency (batch jobs, async workflows), push requests into a durable queue instead of hammering the API. This lets you smooth traffic, survive restarts, and implement priority lanes. Redis + BullMQ (Node) or Redis + RQ (Python) work well. Here’s a minimal Python implementation using SQLite for durability and asyncio for execution:
import sqlite3
import json
import uuid
from dataclasses import dataclass, asdict
from typing import Optional
from contextlib import contextmanager
import asyncio
@dataclass
class QueuedRequest:
id: str
payload: dict
priority: int = 0 # lower = higher priority
created_at: float = 0
attempts: int = 0
max_attempts: int = 3
status: str = "pending" # pending, running, completed, failed
result: Optional[dict] = None
error: Optional[str] = None
class DurableRequestQueue:
def __init__(self, db_path: str = "request_queue.db"):
self.db_path = db_path
self._init_db()
self._worker_task: Optional[asyncio.Task] = None
self._limiter: Optional[TokenAwareLimiter] = None
def _init_db(self):
with self._conn() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS requests (
id TEXT PRIMARY KEY,
payload TEXT NOT NULL,
priority INTEGER DEFAULT 0,
created_at REAL NOT NULL,
attempts INTEGER DEFAULT 0,
max_attempts INTEGER DEFAULT 3,
status TEXT DEFAULT 'pending',
result TEXT,
error TEXT
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_status_priority ON requests(status, priority, created_at)")
@contextmanager
def _conn(self):
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def enqueue(self, payload: dict, priority: int = 0, max_attempts: int = 3) -> str:
req_id = str(uuid.uuid4())
req = QueuedRequest(
id=req_id,
payload=payload,
priority=priority,
created_at=time.time(),
max_attempts=max_attempts
)
with self._conn() as conn:
conn.execute("""
INSERT INTO requests (id, payload, priority, created_at, attempts, max_attempts, status)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (req.id, json.dumps(req.payload), req.priority, req.created_at,
req.attempts, req.max_attempts, req.status))
conn.commit()
return req_id
def _claim_next(self) -> Optional[QueuedRequest]:
with self._conn() as conn:
row = conn.execute("""
SELECT * FROM requests
WHERE status = 'pending'
ORDER BY priority ASC, created_at ASC
LIMIT 1
""").fetchone()
if not row:
return None
req = QueuedRequest(**dict(row))
conn.execute("UPDATE requests SET status = 'running', attempts = attempts + 1 WHERE id = ?", (req.id,))
conn.commit()
return req
def _complete(self, req: QueuedRequest, result: dict):
with self._conn() as conn:
conn.execute("""
UPDATE requests SET status = 'completed', result = ? WHERE id = ?
""", (json.dumps(result), req.id))
conn.commit()
def _fail(self, req: QueuedRequest, error: str):
with self._conn() as conn:
if req.attempts >= req.max_attempts:
conn.execute("UPDATE requests SET status = 'failed', error = ? WHERE id = ?", (error, req.id))
else:
conn.execute("UPDATE requests SET status = 'pending', error = ? WHERE id = ?", (error, req.id))
conn.commit()
async def start_worker(self, limiter: TokenAwareLimiter, poll_interval: float = 1.0):
self._limiter = limiter
self._worker_task = asyncio.create_task(self._worker_loop(poll_interval))
async def _worker_loop(self, poll_interval: float):
while True:
req = self._claim_next()
if req:
try:
response = await self._limiter.request_with_backoff(**req.payload)
result = {"status_code": response.status_code, "body": response.json()}
self._complete(req, result)
except Exception as e:
self._fail(req, str(e))
else:
await asyncio.sleep(poll_interval)
def get_status(self, req_id: str) -> Optional[QueuedRequest]:
with self._conn() as conn:
row = conn.execute("SELECT * FROM requests WHERE id = ?", (req_id,)).fetchone()
return QueuedRequest(**dict(row)) if row else NoneVerify: Enqueue 100 requests with varying priorities. Start the worker. Confirm requests are processed in priority order, retries happen on failure, and the SQLite database survives a process restart with no lost requests.
Step 5: Monitor rate limit headers in real time and alert
You cannot avoid Claude rate limits if you don’t know you’re approaching them. Instrument every request to emit metrics: requests_remaining_pct, tokens_remaining_pct, time_to_reset_seconds. Alert when remaining drops below 20% or when reset is more than 30 seconds away.
from prometheus_client import Gauge, Histogram, Counter
import time
RATE_LIMIT_REQUESTS_REMAINING_PCT = Gauge(
"anthropic_requests_remaining_pct", "Percentage of request quota remaining", ["model"]
)
RATE_LIMIT_TOKENS_REMAINING_PCT = Gauge(
"anthropic_tokens_remaining_pct", "Percentage of token quota remaining", ["model"]
)
RATE_LIMIT_TIME_TO_RESET = Gauge(
"anthropic_time_to_reset_seconds", "Seconds until rate limit resets", ["model", "limit_type"]
)
REQUEST_LATENCY = Histogram(
"anthropic_request_latency_seconds", "Request latency", ["model", "status"]
)
RATE_LIMIT_HITS = Counter(
"anthropic_rate_limit_hits_total", "Total 429 responses", ["model"]
)
async def record_rate_limit_metrics(snapshot: RateLimitSnapshot, model: str, latency: float, status: int):
if snapshot.requests_limit > 0:
RATE_LIMIT_REQUESTS_REMAINING_PCT.labels(model=model).set(
(snapshot.requests_remaining / snapshot.requests_limit) * 100
)
if snapshot.tokens_limit > 0:
RATE_LIMIT_TOKENS_REMAINING_PCT.labels(model=model).set(
(snapshot.tokens_remaining / snapshot.tokens_limit) * 100
)
for limit_type, reset_header in [("requests", snapshot.reset_requests_at),
("tokens", snapshot.reset_tokens_at)]:
if reset_header:
try:
from datetime import datetime, timezone
dt = datetime.fromisoformat(reset_header.replace("Z", "+00:00"))
reset_ts = dt.timestamp()
wait = max(0, reset_ts - time.time())
RATE_LIMIT_TIME_TO_RESET.labels(model=model, limit_type=limit_type).set(wait)
except Exception:
pass
REQUEST_LATENCY.labels(model=model, status=str(status)).observe(latency)
if status == 429:
RATE_LIMIT_HITS.labels(model=model).inc()Verify: Generate load until you hit a 429. Check your Prometheus/Grafana dashboard: anthropic_requests_remaining_pct should approach 0, anthropic_time_to_reset_seconds should show the reset window, and anthropic_rate_limit_hits_total should increment. Set an alert on anthropic_requests_remaining_pct < 20 for 5 minutes.
Step 6: Implement provider fallback for critical paths
When your primary provider is rate-limited or degraded, fail over to an equivalent model on another provider. This requires a unified interface that normalizes request/response formats across providers. The key is routing directives: your client specifies preferences (e.g., “prefer Claude 3.5 Sonnet, fallback to GPT-4o”), and the gateway handles the translation.
from enum import Enum
from typing import Literal
import httpx
class Provider(str, Enum):
ANTHROPIC = "anthropic"
OPENAI = "openai"
# Add others as needed
MODEL_ALIASES = {
"claude-3-5-sonnet": {
Provider.ANTHROPIC: "claude-3-5-sonnet-20241022",
Provider.OPENAI: "gpt-4o-2024-08-06", # rough equivalent
},
"claude-3-opus": {
Provider.ANTHROPIC: "claude-3-opus-20240229",
Provider.OPENAI: "gpt-4-turbo-2024-04-09",
},
}
class UnifiedLLMClient:
def __init__(self,
anthropic_key: str,
openai_key: str,
preferred_provider: Provider = Provider.ANTHROPIC,
fallback_providers: list[Provider] = None):
self.clients = {
Provider.ANTHROPIC: httpx.AsyncClient(
base_url="https://api.anthropic.com/v1",
headers={"x-api-key": anthropic_key, "anthropic-version": "2023-06-01"},
timeout=60.0
),
Provider.OPENAI: httpx.AsyncClient(
base_url="https://api.openai.com/v1",
headers={"Authorization": f"Bearer {openai_key}"},
timeout=60.0
),
}
self.preferred = preferred_provider
self.fallbacks = fallback_providers or [p for p in Provider if p != preferred_provider]
self.limiters = {
Provider.ANTHROPIC: TokenAwareLimiter(self.clients[Provider.ANTHROPIC]),
Provider.OPENAI: TokenAwareLimiter(self.clients[Provider.OPENAI]), # adapt for OpenAI headers
}
def _translate_to_anthropic(self, payload: dict) -> dict:
"""Convert OpenAI-style payload to Anthropic format."""
return {
"model": payload["model"],
"messages": payload["messages"],
"system": payload.get("system", ""),
"max_tokens": payload.get("max_tokens", 4096),
"temperature": payload.get("temperature", 0.7),
}
def _translate_to_openai(self, payload: dict) -> dict:
"""Convert Anthropic-style payload to OpenAI format."""
messages = []
if payload.get("system"):
messages.append({"role": "system", "content": payload["system"]})
messages.extend(payload["messages"])
return {
"model": payload["model"],
"messages": messages,
"max_tokens": payload.get("max_tokens", 4096),
"temperature": payload.get("temperature", 0.7),
}
async def complete(self,
model_alias: str,
messages: list[dict],
system: str = "",
max_tokens: int = 4096,
temperature: float = 0.7,
**kwargs) -> httpx.Response:
"""Try preferred provider, then fallbacks in order."""
providers_to_try = [self.preferred] + self.fallbacks
last_error = None
for provider in providers_to_try:
if model_alias not in MODEL_ALIASES:
raise ValueError(f"Unknown model alias: {model_alias}")
actual_model = MODEL_ALIASES[model_alias].get(provider)
if not actual_model:
continue # provider doesn't support this alias
limiter = self.limiters[provider]
client = self.clients[provider]
if provider == Provider.ANTHROPIC:
payload = {
"model": actual_model,
"messages": messages,
"system": system,
"max_tokens": max_tokens,
"temperature": temperature,
"api_key": client.headers["x-api-key"],
}
elif provider == Provider.OPENAI:
payload = {
"model": actual_model,
"messages": [{"role": "system", "content": system}] + messages if system else messages,
"max_tokens": max_tokens,
"temperature": temperature,
}
try:
if provider == Provider.ANTHROPIC:
response = await limiter.request_with_backoff(**payload)
else:
# OpenAI path — implement similar backoff for their headers
response = await client.post("/chat/completions", json=payload)
if response.status_code == 429:
last_error = f"{provider.value} rate limited"
continue
elif response.status_code >= 500:
last_error = f"{provider.value} server error: {response.status_code}"
continue
# Success — normalize response format if needed
return response
except Exception as e:
last_error = f"{provider.value} error: {e}"
continue
raise RuntimeError(f"All providers failed. Last error: {last_error}")Verify: Set your Anthropic key to a quota-exhausted account. Send a request with model_alias="claude-3-5-sonnet". Confirm the response comes from OpenAI’s GPT-4o with a 200 status, and the latency is reasonable. Check that the anthropic_rate_limit_hits_total metric increments for the Anthropic attempt.
Step 7: Optimize token usage to stretch your quota
The most effective way to avoid Claude rate limits is to use fewer tokens. Three high-leverage techniques:
- Compress prompts — Remove redundant instructions, use structured formats (JSON schemas instead of prose), and drop few-shot examples once the model is reliable.
- Enable prompt caching — Anthropic supports
cache_control: {"type": "ephemeral"}on system prompts and long context blocks. This can reduce token consumption by 90% for repeated prefixes. - Stream and truncate — For long generations, stream the response and stop early if you have what you need.
def build_cached_system_prompt(base_instructions: str, dynamic_context: str) -> list[dict]:
"""Structure system prompt for Anthropic prompt caching."""
return [
{
"type": "text",
"text": base_instructions,
"cache_control": {"type": "ephemeral"} # cached across requests
},
{
"type": "text",
"text": dynamic_context,
# no cache_control — this changes per request
}
]
async def stream_with_early_stop(limiter: TokenAwareLimiter,
messages: list[dict],
system: list[dict],
max_tokens: int,
stop_sequences: list[str] = None,
**kwargs) -> str:
"""Stream response and stop at first stop sequence."""
payload = {
"model": kwargs.get("model", "claude-3-5-sonnet-20241022"),
"messages": messages,
"system": system,
"max_tokens": max_tokens,
"stream": True,
"stop_sequences": stop_sequences or [],
}
response = await limiter.client.post(
"https://api.anthropic.com/v1/messages",
json=payload,
headers={
"x-api-key": kwargs.get("api_key"),
"anthropic-version": "2023-06-01",
"content-type": "application/json",
},
timeout=None # streaming needs no timeout
)
if response.status_code != 200:
raise RuntimeError(f"Stream failed: {response.status_code} {await response.aread()}")
accumulated = []
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
data = line[6:]
if data == "[DONE]":
break
try:
import json
event = json.loads(data)
if event.get("type") == "content_block_delta":
text = event.get("delta", {}).get("text", "")
accumulated.append(text)
# Check stop sequences
full_text = "".join(accumulated)
if any(seq in full_text for seq in (stop_sequences or [])):
break
except json.JSONDecodeError:
continue
return "".join(accumulated)Verify: Send 100 requests with a 2,000-token cached system prompt and 500-token dynamic context. Check the anthropic-ratelimit-tokens-remaining header — the first request should consume ~2,500 tokens; subsequent requests should consume only ~500 tokens (the uncached portion). Confirm the cached tokens are billed at the discounted cache rate on your invoice.
How to verify the whole system works end to end
Run this integration test against a staging environment:
async def integration_test():
# Setup
anthropic_key = os.getenv("ANTHROPIC_API_KEY")
openai_key = os.getenv("OPENAI_API_KEY")
client = UnifiedLLMClient(anthropic_key, openai_key)
queue = DurableRequestQueue()
await queue.start_worker(client.limiters[Provider.ANTHROPIC])
# Enqueue burst of 200 requests
for i in range(200):
queue.enqueue({
"model_alias": "claude-3-5-sonnet",
"messages": [{"role": "user", "content": f"Count to 10. Request {i}"}],
"max_tokens": 100,
}, priority=0 if i < 50 else 1) # first 50 are high priority
# Wait for completion
await asyncio.sleep(120)
# Verify
completed = 0
failed = 0
rate_limited = 0
with queue._conn() as conn:
rows = conn.execute("SELECT status, error, result FROM requests").fetchall()
for row in rows:
if row["status"] == "completed":
completed += 1
result = json.loads(row["result"])
if result["status_code"] == 429:
rate_limited += 1
elif row["status"] == "failed":
failed += 1
print(f"Completed: {completed}, Failed: {failed}, Rate limited responses: {rate_limited}")
assert failed == 0, "No requests should fail permanently"
assert rate_limited == 0, "No 429s should reach the application layer"
print("Integration test PASSED")Success criteria: Zero failed requests, zero 429 responses reaching your application logic, high-priority requests complete before low-priority ones, and the fallback provider is never invoked unless Anthropic is genuinely unavailable (not just rate-limited).
The layers compound: header-aware backoff handles transient spikes, token-aware concurrency prevents self-inflicted limit exhaustion, the durable queue absorbs bursts, metrics give you visibility, fallback providers handle sustained outages, and token optimization raises the ceiling for everything. Deploy them incrementally — start with Steps 1 and 2, verify, then add the queue, then monitoring, then fallback. Each layer reduces the blast radius of the next rate limit event.