llm-gateway

Production-ready proxy for the Anthropic API. Drop it in front of any code that calls the API and get rate limiting, automatic retry with backoff, per-call cost tracking, and structured JSON logs — all without changing your existing call sites.

What it does

Feature	Detail
Rate limiting	Token bucket (configurable RPS + burst). Thread-safe — safe to share across threads
Retry	Exponential backoff with full jitter on 429, 500, 502, 503, 529. Non-retryable 4xx errors are re-raised immediately
Cost tracking	Per-call USD cost including cache write/read tokens. Running totals in /metrics
Structured logs	One JSON line per request: req_id, model, latency_ms, token breakdown, cost_usd
HTTP proxy	FastAPI app — POST /v1/messages, GET /health, GET /metrics

Install

pip install -e .

Quickstart

As a library (zero HTTP overhead):

import anthropic
from llmgateway import Gateway

gw = Gateway(client=anthropic.Anthropic(), rps=5)

response = gw.call({
    "model": "claude-sonnet-4-6",
    "max_tokens": 256,
    "messages": [{"role": "user", "content": "What is 12 * 37?"}],
})
print(response["content"][0]["text"])

print(gw.metrics())

{"req_id": "a3f2b1c0", "model": "claude-sonnet-4-6", "latency_ms": 842.3,
 "tokens": {"input": 18, "output": 9, "cache_read": 0, "cache_write": 0},
 "cost_usd": 0.000189}

As an HTTP server (drop-in API proxy):

llm-gateway --host 0.0.0.0 --port 8080 --rps 10
# or
python -m llmgateway --port 8080

curl http://localhost:8080/v1/messages \
  -H "Content-Type: application/json" \
  -d '{"model":"claude-sonnet-4-6","max_tokens":64,"messages":[{"role":"user","content":"ping"}]}'

curl http://localhost:8080/metrics

{
  "requests": 42,
  "errors": 1,
  "error_rate": 0.0238,
  "total_cost_usd": 0.008714,
  "total_input_tokens": 18430,
  "total_output_tokens": 3214
}

Environment variables

Variable	Default	Description
ANTHROPIC_API_KEY	—	Required. Read by the Anthropic client
HOST	127.0.0.1	Bind address
PORT	8080	Listen port
RPS	10	Sustained rate limit

Architecture

             ┌─────────────────────────────────┐
             │           Gateway               │
request ───► │  TokenBucketLimiter.acquire()   │
             │         ↓                       │
             │  with_retry(client.messages...) │──► Anthropic API
             │         ↓                       │◄──
             │  call_cost_usd(usage)           │
             │  _emit_log(JSON)                │
             └─────────────────────────────────┘
                         ↓
                   response dict

Gateway — thin orchestration layer. Stateful (metrics accumulate). Thread-safe via the limiter's lock.

TokenBucketLimiter — classic token bucket. Tokens refill continuously at rps/sec, capped at burst. acquire(block=False) is non-blocking for backpressure scenarios.

with_retry — full jitter backoff: delay = min(base * 2^attempt, max) + uniform(0, jitter). Full jitter outperforms equal/decorrelated jitter under sustained load (AWS paper, 2015).

call_cost_usd — exact pricing for Opus 4.8, Sonnet 4.6, Haiku 4.5 including cache write (125% base) and cache read (10% base). Unknown models fall back to Sonnet pricing.

Testing

pip install -e ".[dev]"
pytest -q

All tests use a mock client — no API key needed.

............................
28 passed in 0.08s