JWT Auth and API Key Management in FastAPI: A Production Guide

Authentication is the first thing you build and the last thing you want to debug in production. This guide covers the full authentication layer for a FastAPI AI API: JWT user auth, API key issuance, per-key rate limiting, and the security details that most tutorials skip.

Two authentication systems, one application

AI APIs typically need two kinds of auth:

1. JWT tokens for user authentication (login flows, dashboards)
2. API keys for programmatic access (SDK calls, integrations)

Both need to coexist, and both need to integrate with your rate limiting and billing systems.

JWT implementation

Use python-jose for JWT operations and passlib for password hashing:

# app/auth/jwt.py
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

SECRET_KEY = settings.JWT_SECRET_KEY
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = timedelta(minutes=30)
REFRESH_TOKEN_EXPIRE = timedelta(days=7)

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(user_id: str) -> str:
    payload = {
        "sub": user_id,
        "type": "access",
        "exp": datetime.utcnow() + ACCESS_TOKEN_EXPIRE,
        "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(user_id: str) -> str:
    payload = {
        "sub": user_id,
        "type": "refresh",
        "exp": datetime.utcnow() + REFRESH_TOKEN_EXPIRE,
        "iat": datetime.utcnow(),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

async def verify_token(token: str, expected_type: str = "access") -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != expected_type:
            raise HTTPException(status_code=401, detail="Invalid token type")
        return payload["sub"]
    except JWTError:
        raise HTTPException(status_code=401, detail="Could not validate credentials")

API key system

API keys need to be secure, searchable, and revokable. Store a hash of the key, not the key itself:

# app/models/api_key.py
import secrets
import hashlib

class APIKey(Base):
    __tablename__ = "api_keys"
    
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
    key_hash: Mapped[str] = mapped_column(unique=True, index=True)
    key_prefix: Mapped[str]  # First 8 chars for display: "kit_live_ab1c..."
    
    owner_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
    name: Mapped[str]  # Human-readable label
    tier: Mapped[str] = mapped_column(default="basic")
    
    rate_limit_per_minute: Mapped[int] = mapped_column(default=60)
    rate_limit_per_day: Mapped[int] = mapped_column(default=5000)
    
    is_active: Mapped[bool] = mapped_column(default=True)
    last_used_at: Mapped[datetime | None]
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

def generate_api_key() -> tuple[str, str, str]:
    """Returns (raw_key, key_hash, key_prefix)"""
    raw = f"kit_live_{secrets.token_urlsafe(32)}"
    hashed = hashlib.sha256(raw.encode()).hexdigest()
    prefix = raw[:12]
    return raw, hashed, prefix

Issuing keys

@router.post("/v1/keys", response_model=APIKeyResponse)
async def issue_api_key(
    body: CreateAPIKeyRequest,
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    raw_key, key_hash, prefix = generate_api_key()
    
    api_key = APIKey(
        key_hash=key_hash,
        key_prefix=prefix,
        owner_id=user.id,
        name=body.name,
        tier=body.tier or "basic",
        rate_limit_per_minute=TIER_LIMITS[body.tier]["per_minute"],
        rate_limit_per_day=TIER_LIMITS[body.tier]["per_day"],
    )
    db.add(api_key)
    await db.commit()
    
    # Return raw key ONCE — never stored, never retrievable again
    return APIKeyResponse(
        id=api_key.id,
        key=raw_key,  # Only time this is shown
        prefix=prefix,
        name=body.name,
    )

Rate limiting via Redis

The @rate_limit decorator uses Redis sliding window counters:

# app/auth/rate_limit.py
import functools
from app.cache import redis_client

def rate_limit(per_minute: int = 60, per_day: int = 5000):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, key: APIKey, **kwargs):
            minute_key = f"rl:{key.id}:min:{int(time.time() // 60)}"
            day_key = f"rl:{key.id}:day:{datetime.utcnow().date()}"
            
            pipe = redis_client.pipeline()
            pipe.incr(minute_key)
            pipe.expire(minute_key, 61)
            pipe.incr(day_key)
            pipe.expire(day_key, 86401)
            results = await pipe.execute()
            
            minute_count, _, day_count, _ = results
            
            if minute_count > per_minute:
                raise HTTPException(
                    status_code=429,
                    headers={"Retry-After": "60", "X-RateLimit-Remaining": "0"},
                    detail=f"Rate limit exceeded: {per_minute}/min",
                )
            if day_count > per_day:
                raise HTTPException(
                    status_code=429,
                    headers={"Retry-After": "86400", "X-RateLimit-Remaining": "0"},
                    detail=f"Daily limit exceeded: {per_day}/day",
                )
            return await func(*args, key=key, **kwargs)
        return wrapper
    return decorator

Protecting routes

@router.post("/v1/chat")
@require_api_key(tier=["basic", "pro"])
@rate_limit(per_minute=60, per_day=5000)
async def chat(
    body: ChatRequest,
    key: APIKey = Depends(get_api_key),
):
    # key is validated, rate checked, tier verified
    return await process_chat(body, key)

Security details most guides skip

1. Constant-time comparison for API keys

import hmac
# Wrong — vulnerable to timing attacks
if stored_hash == provided_hash:
    pass
# Right
if hmac.compare_digest(stored_hash, provided_hash):
    pass

2. Update last_used_at without blocking the request

# Don't await this — fire and forget
asyncio.create_task(
    update_key_last_used(key.id)
)

3. Key rotation endpoint

Always provide a way to rotate keys without downtime — issue new key before revoking old.

FastAPI AI Kit implements all of this in app/auth/ — JWT, API keys, rate limiting, secure storage, and the endpoints. You configure tier limits in settings.py and the rest works.