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

CitizenApp engineers used FastAPI's dependency injection to isolate Anthropic Claude API keys and rate limits per tenant, replacing a fragile global key setup that throttled all customers when one hit quota. The system composes per-request dependencies for tenant config, rate-limit buckets, and Claude clients, avoiding middleware spaghetti and context-var leaks.

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: python 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: python 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: python 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} 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: python 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 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