When CitizenApp hit 15 tenants, I realized our single global Claude API key was a ticking time bomb. One customer's agentic loop burning through their quota would throttle everyone else. Worse, we had no way to enforce per-tenant rate limits without adding middleware spaghetti that would make debugging a nightmare.
The fix? Lean into FastAPI's dependency injection system to make tenant-specific Claude clients and rate-limit buckets first-class citizens. No globals, no thread locks, no "who's using the API key right now?" detective work.
Middleware runs once per request, which means you'd have to either:
I've been burned by this. We had a get_current_tenant()
middleware that set request.state.tenant_id
, but then handlers had to manually fetch the API key and pass it around. When we added background tasks that queried Claude, the entire pattern fell apart—context vars leaked, rate limits weren't enforced, and debugging which tenant was which took hours.
FastAPI's Depends system solves this cleanly: dependencies are resolved per-request (or per-dependency cache if you use use_cache=True
), and they compose naturally. Your handler doesn't care how it gets a Claude client—it just declares what it needs.
Let's start with the data layer. You need a way to fetch tenant configuration and manage rate limits:
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.orm import Session
from datetime import datetime, timedelta
class Tenant(Base):
__tablename__ = "tenants"
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
anthropic_api_key = Column(String) # Encrypted in production
max_requests_per_minute = Column(Integer, default=60)
preferred_model = Column(String, default="claude-3-5-sonnet-20241022")
class RateLimitBucket:
"""In-memory rate limit tracker. Use Redis for distributed deployments."""
def __init__(self, max_requests: int, window_seconds: int = 60):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests: list[datetime] = []
def is_allowed(self) -> bool:
now = datetime.utcnow()
cutoff = now - timedelta(seconds=self.window_seconds)
self.requests = [req for req in self.requests if req > cutoff]
if len(self.requests) < self.max_requests:
self.requests.append(now)
return True
return False
Now the dependency providers:
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthCredential
from sqlalchemy.orm import Session
import anthropic
from functools import lru_cache
security = HTTPBearer()
def get_db() -> Session:
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_tenant_id(credentials: HTTPAuthCredential = Depends(security)) -> int:
"""Extract and validate the tenant from JWT or API key header."""
try:
payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=["HS256"])
tenant_id = payload.get("tenant_id")
if not tenant_id:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
return tenant_id
except jwt.InvalidTokenError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
def get_tenant(
tenant_id: int = Depends(get_tenant_id),
db: Session = Depends(get_db)
) -> Tenant:
"""Fetch the tenant record. This runs once per request."""
tenant = db.query(Tenant).filter(Tenant.id == tenant_id).first()
if not tenant:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return tenant
_rate_limit_buckets: dict[int, RateLimitBucket] = {}
_claude_clients: dict[int, anthropic.Anthropic] = {}
def get_claude_client(tenant: Tenant = Depends(get_tenant)) -> anthropic.Anthropic:
"""
Get or create a Claude client for this tenant.
Reused within the request if other dependencies need it.
"""
if tenant.id not in _claude_clients:
_claude_clients[tenant.id] = anthropic.Anthropic(
api_key=tenant.anthropic_api_key
)
return _claude_clients[tenant.id]
def get_rate_limit_bucket(tenant: Tenant = Depends(get_tenant)) -> RateLimitBucket:
"""
Get or create the rate-limit bucket for this tenant.
Separate from Claude client so you can inject one without the other if needed.
"""
if tenant.id not in _rate_limit_buckets:
_rate_limit_buckets[tenant.id] = RateLimitBucket(
max_requests=tenant.max_requests_per_minute
)
return _rate_limit_buckets[tenant.id]
def check_rate_limit(bucket: RateLimitBucket = Depends(get_rate_limit_bucket)) -> None:
"""Dependency that enforces the rate limit. Use in handlers that call Claude."""
if not bucket.is_allowed():
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Rate limit exceeded for this tenant"
)
Now your handlers are clean and testable:
from fastapi import FastAPI, Depends
from pydantic import BaseModel
app = FastAPI()
class ChatRequest(BaseModel):
message: str
@app.post("/chat")
async def chat(
req: ChatRequest,
client: anthropic.Anthropic = Depends(get_claude_client),
tenant: Tenant = Depends(get_tenant),
_: None = Depends(check_rate_limit), # Rate limit is checked first
):
"""
The handler only declares what it needs.
FastAPI wires up the tenant, validates their rate limit, and gives us a pre-configured Claude client.
"""
response = client.messages.create(
model=tenant.preferred_model,
max_tokens=1024,
messages=[{"role": "user", "content": req.message}]
)
return {"response": response.content[0].text}
@app.post("/batch-analyze")
async def batch_analyze(
files: list[UploadFile],
client: anthropic.Anthropic = Depends(get_claude_client),
bucket: RateLimitBucket = Depends(get_rate_limit_bucket),
tenant: Tenant = Depends(get_tenant),
):
"""
You can also use the bucket directly if you need fine-grained control.
E.g., consume multiple tokens per request.
"""
results = []
for file in files:
if not bucket.is_allowed():
return {"error": "Rate limit exceeded mid-batch", "processed": len(results)}
content = await file.read()
response = client.messages.create(
model=tenant.preferred_model,
max_tokens=512,
messages=[{"role": "user", "content": f"Analyze: {content.decode()}"}]
)
results.append(response.content[0].text)
return {"results": results}
get_claude_client
depends on get_tenant
, which depends on get_tenant_id
. You can test each layer independently.I initially used lru_cache
on get_tenant()
to avoid DB hits. Don't. If a tenant's API key rotates mid-day, cached tenants still have the old key. Instead:
@lru_cache(maxsize=128)
def get_tenant(tenant_id: int, db: Session):
return db.query(Tenant).filter(Tenant.id == tenant_id).first()
def get_tenant(tenant_id: int = Depends(get_tenant_id), db: Session = Depends(get_db)):
return db.query(Tenant).filter(Tenant.id == tenant_id).first()
The DB query is cheap. Stale credentials are expensive.
For distributed deployments with multiple FastAPI instances, replace in-memory buckets with Redis:
python
import redis
from datetime import datetime, timedelta
def get_rate_limit_bucket(tenant: Tenant = Depends(get_tenant)) -> int:
"""Return the tenant ID; use Redis in the handler."""
return tenant.id
@app.post("/chat")
async def chat(
req: ChatRequest,
client: anthropic.Anthropic = Depends