FastAPI Dependency Injection for Anthropic Claude: Isolating API Keys and Rate Limits Per Tenant

FastAPI Dependency Injection for Anthropic Claude: Isolating API Keys and Rate Limits Per Tenant 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: Parse the tenant ID from the request, look up their key, then store it somewhere accessible (request state, context vars, thread-local storage) Hope that concurrent requests don't collide when accessing shared rate-limit buckets 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: # models.py 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: # dependencies.py 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: # Standard FastAPI DB dependency 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.""" # In reality, decode your JWT here 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 # Global cache for rate-limit buckets and Claude clients # Keys are tenant IDs. In production, use Redis. _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: # routes.py 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} Isolation: Each tenant's API key and rate limit are independent. One tenant's quota exhaustion doesn't touch another's. Composability: Dependencies depend on other dependencies. get_claude_client depends on get_tenant, which depends on get_tenant_id. You can test each layer independently. Reusability: If 10 handlers need Claude, they all get the same tenant-specific client without duplication. Async-safe: FastAPI resolves dependencies per-request. No thread-local trickery, no accidental state sharing between concurrent requests. 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: # Bad @lru_cache(maxsize=128) def get_tenant(tenant_id: int, db: Session): return db.query(Tenant).filter(Tenant.id == tenant_id).first() # Good 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