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.

Why Not Middleware or Global State?

Middleware runs once per request, which means you'd have to either:

1. Parse the tenant ID from the request, look up their key, then store it somewhere accessible (request state, context vars, thread-local storage)
2. 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.

The Setup: Tenant-Aware Dependency Providers

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"
        )

Using It in Handlers

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}

Why This Works

1. Isolation: Each tenant's API key and rate limit are independent. One tenant's quota exhaustion doesn't touch another's.
2. 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.
3. Reusability: If 10 handlers need Claude, they all get the same tenant-specific client without duplication.
4. Async-safe: FastAPI resolves dependencies per-request. No thread-local trickery, no accidental state sharing between concurrent requests.

Gotcha: Caching Pitfalls

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.

Scaling to Redis

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