Lesson 4: Rate Limiting, Quotas & Multi-Tenancy

Rate limits are a shared resource problem: your application competes for API capacity with itself (multiple threads, users, microservices) and the provider's other customers. Client-side rate limiting prevents you from hitting provider limits and gives you per-tenant cost control.

Anthropic API Rate Limit Types

Default limit:50 RPM (free tier haiku)

When hit:HTTP 429 with Retry-After header. All inflight requests for that minute are blocked until the window resets.

Client control:Token bucket: allow burst of N, refill at rate r/s. Requests block rather than fail.

import time, threading
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    capacity: int      # Max burst size (e.g., 10 requests)
    rate: float        # Refill rate (e.g., 50 per 60s = 0.833/s)
    _tokens: float = field(init=False)
    _last_refill: float = field(default_factory=time.monotonic, init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, init=False)

    def __post_init__(self):
        self._tokens = float(self.capacity)

    def acquire(self, timeout: float = 30.0) -> bool:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            with self._lock:
                elapsed = time.monotonic() - self._last_refill
                self._tokens = min(self.capacity, self._tokens + elapsed * self.rate)
                self._last_refill = time.monotonic()
                if self._tokens >= 1:
                    self._tokens -= 1
                    return True
            time.sleep(1.0 / self.rate)
        return False

# 50 RPM, allow burst of 10
rpm_limiter = TokenBucket(capacity=10, rate=50/60)

def call_llm(prompt: str) -> str:
    if not rpm_limiter.acquire(timeout=60):
        raise TimeoutError("Could not acquire RPM slot within 60s")
    return client.messages.create(...).content[0].text

Track tokens, not requests

Requests vary wildly in cost. A 1-token request and a 10K-token request both count as 1 RPM but have very different costs. Track input+output tokens for meaningful quota enforcement.

Soft limits before hard limits

Alert tenants at 80% of quota usage. Hard-cut at 100%. A surprised user who was never warned is an angry user — give them the chance to upgrade.

Expose quota status in API headers

Return X-Quota-Used: 45000, X-Quota-Limit: 50000 in every response. Clients can implement their own throttling before hitting the hard limit.

Pre-flight token estimation prevents quota overshoot.Estimate input tokens before the call using len(prompt) / 4(rough approximation). If the estimate would exceed quota, reject early rather than discovering overshoot after the API call — you've already paid for those tokens.

import time, threading from dataclasses import dataclass, field @dataclass class TokenBucket: capacity: int # Max burst size (e.g., 10 requests) rate: float # Refill rate (e.g., 50 per 60s = 0.833/s) _tokens: float = field(init=False) _last_refill: float = field(default_factory=time.monotonic, init=False) _lock: threading.Lock = field(default_factory=threading.Lock, init=False) def __post_init__(self): self._tokens = float(self.capacity) def acquire(self, timeout: float = 30.0) -> bool: deadline = time.monotonic() + timeout while time.monotonic() < deadline: with self._lock: elapsed = time.monotonic() - self._last_refill self._tokens = min(self.capacity, self._tokens + elapsed * self.rate) self._last_refill = time.monotonic() if self._tokens >= 1: self._tokens -= 1 return True time.sleep(1.0 / self.rate) return False # 50 RPM, allow burst of 10 rpm_limiter = TokenBucket(capacity=10, rate=50/60) def call_llm(prompt: str) -> str: if not rpm_limiter.acquire(timeout=60): raise TimeoutError("Could not acquire RPM slot within 60s") return client.messages.create(...).content[0].text