How to scope API keys with rate limits and allowlists

Scoping API keys with rate limits and allowlists is the difference between a prototype that works and a production system that survives. Without it, one runaway script or compromised key can exhaust your budget, trigger provider bans, or leak data across tenants. This guide walks through designing and enforcing scopes end to end — key hierarchy, per-scope quotas, network and model allowlists, rotation, and observability — so you can ship multi-tenant LLM features without losing sleep.

Step 1: Define your scope model

Before writing code, decide what a “scope” represents in your system. Common dimensions:

• Tenant or workspace — isolates customer A from customer B
• Environment — dev, staging, prod with different budgets
• Feature or product line — chat, embeddings, fine-tuning jobs
• Identity tier — free, pro, enterprise with different limits

Model each scope as a distinct key prefix or metadata tag. A clean pattern: sk_{scope}_{random}. For example, sk_prod_acme_corp_abc123 or sk_dev_feature_chat_xyz789. The prefix lets you route, meter, and revoke by scope without a database lookup on every request.

# scopes.py
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import secrets

class ScopeType(Enum):
    TENANT = "tenant"
    ENVIRONMENT = "env"
    FEATURE = "feature"
    TIER = "tier"

@dataclass(frozen=True)
class KeyScope:
    scope_type: ScopeType
    scope_id: str
    metadata: dict  # arbitrary tags: {"region": "us-east", "team": "platform"}

    def prefix(self) -> str:
        return f"sk_{self.scope_type.value}_{self.scope_id}"

def generate_key(scope: KeyScope, entropy_bytes: int = 16) -> str:
    suffix = secrets.token_urlsafe(entropy_bytes)
    return f"{scope.prefix()}_{suffix}"

# Usage
prod_scope = KeyScope(ScopeType.TENANT, "acme_corp", {"tier": "enterprise", "region": "us-east"})
api_key = generate_key(prod_scope)
# -> "sk_tenant_acme_corp_A1b2C3d4E5f6G7h8"

Verify: Parse a generated key back into its scope components with a regex. No false positives.

import re

KEY_PATTERN = re.compile(r"^sk_(?P<type>\w+)_(?P<id>[^_]+)_(?P<suffix>.+)$")

def parse_key(key: str) -> Optional[KeyScope]:
    m = KEY_PATTERN.match(key)
    if not m:
        return None
    return KeyScope(ScopeType(m.group("type")), m.group("id"), {})

Step 2: Store keys with scoped metadata

Persist each key with its scope, rate limit policy, allowlists, and status. Use a table keyed by the full key hash (never store plaintext). Include a key_hash column for lookups, plus scope_type, scope_id, rate_limit_rpm, rate_limit_tpm, ip_allowlist, domain_allowlist, model_allowlist, status, created_at, revoked_at.

-- migration: 001_create_api_keys.sql
CREATE TABLE api_keys (
    key_hash        BYTEA PRIMARY KEY,          -- SHA256 of full key
    key_prefix      TEXT NOT NULL,              -- "sk_tenant_acme_corp" for prefix queries
    scope_type      TEXT NOT NULL,              -- "tenant", "env", "feature", "tier"
    scope_id        TEXT NOT NULL,              -- "acme_corp", "prod", "chat", "enterprise"
    metadata        JSONB NOT NULL DEFAULT '{}',
    rate_limit_rpm  INTEGER NOT NULL DEFAULT 60,   -- requests per minute
    rate_limit_tpm  INTEGER NOT NULL DEFAULT 100000, -- tokens per minute
    ip_allowlist    TEXT[] NOT NULL DEFAULT '{}',  -- CIDR blocks
    domain_allowlist TEXT[] NOT NULL DEFAULT '{}', -- e.g., "app.example.com"
    model_allowlist TEXT[] NOT NULL DEFAULT '{}',  -- e.g., "gpt-4o", "claude-3.5-sonnet"
    status          TEXT NOT NULL DEFAULT 'active', -- active, revoked, expired
    created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
    revoked_at      TIMESTAMPTZ,
    expires_at      TIMESTAMPTZ
);

CREATE INDEX idx_api_keys_prefix ON api_keys (key_prefix);
CREATE INDEX idx_api_keys_scope ON api_keys (scope_type, scope_id);

Hash the key on ingest. Never log the full key.

# key_store.py
import hashlib
import hmac
from datetime import datetime, timezone
from typing import Optional
import asyncpg

class KeyStore:
    def __init__(self, pool: asyncpg.Pool):
        self.pool = pool

    @staticmethod
    def hash_key(key: str) -> bytes:
        return hashlib.sha256(key.encode()).digest()

    @staticmethod
    def prefix(key: str) -> str:
        # "sk_tenant_acme_corp_A1b2..." -> "sk_tenant_acme_corp"
        parts = key.split("_", 3)
        return "_".join(parts[:3]) if len(parts) >= 3 else key

    async def create_key(self, key: str, scope: KeyScope, policy: dict) -> None:
        key_hash = self.hash_key(key)
        prefix = self.prefix(key)
        async with self.pool.acquire() as conn:
            await conn.execute("""
                INSERT INTO api_keys (key_hash, key_prefix, scope_type, scope_id, metadata,
                                      rate_limit_rpm, rate_limit_tpm,
                                      ip_allowlist, domain_allowlist, model_allowlist)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
            """, key_hash, prefix, scope.scope_type.value, scope.scope_id,
                scope.metadata,
                policy.get("rpm", 60),
                policy.get("tpm", 100_000),
                policy.get("ip_allowlist", []),
                policy.get("domain_allowlist", []),
                policy.get("model_allowlist", []))

    async def lookup(self, key: str) -> Optional[dict]:
        key_hash = self.hash_key(key)
        async with self.pool.acquire() as conn:
            row = await conn.fetchrow("""
                SELECT * FROM api_keys WHERE key_hash = $1 AND status = 'active'
            """, key_hash)
            return dict(row) if row else None

    async def revoke_by_scope(self, scope_type: ScopeType, scope_id: str) -> int:
        async with self.pool.acquire() as conn:
            result = await conn.execute("""
                UPDATE api_keys
                SET status = 'revoked', revoked_at = now()
                WHERE scope_type = $1 AND scope_id = $2 AND status = 'active'
            """, scope_type.value, scope_id)
            return int(result.split()[-1])  # "UPDATE 3" -> 3

Verify: Insert a key, look it up by hash, confirm status = 'active'. Revoke by scope, confirm affected row count matches.

Step 3: Enforce per-scope rate limits

Rate limiting must happen before the request reaches the upstream provider. Use a sliding-window algorithm backed by Redis for low latency and horizontal scaling. Track two counters per scope: requests per minute (RPM) and tokens per minute (TPM). Token counts come from the request body (estimated) and response (actual).

# rate_limiter.py
import time
from dataclasses import dataclass
from typing import Optional
import redis.asyncio as redis

@dataclass
class RateLimitResult:
    allowed: bool
    rpm_remaining: int
    tpm_remaining: int
    retry_after_ms: Optional[int] = None

class SlidingWindowRateLimiter:
    def __init__(self, client: redis.Redis, window_seconds: int = 60):
        self.client = client
        self.window = window_seconds

    async def check_and_increment(
        self,
        scope_key: str,
        rpm_limit: int,
        tpm_limit: int,
        estimated_tokens: int = 0
    ) -> RateLimitResult:
        now = time.time()
        window_start = now - self.window
        rpm_key = f"rl:rpm:{scope_key}"
        tpm_key = f"rl:tpm:{scope_key}"

        async with self.client.pipeline(transaction=True) as pipe:
            # Remove expired entries
            pipe.zremrangebyscore(rpm_key, 0, window_start)
            pipe.zremrangebyscore(tpm_key, 0, window_start)
            # Count current
            pipe.zcard(rpm_key)
            pipe.zcard(tpm_key)
            results = await pipe.execute()

        current_rpm = results[2]
        current_tpm = results[3]

        if current_rpm >= rpm_limit:
            # Get oldest entry to calculate retry-after
            oldest = await self.client.zrange(rpm_key, 0, 0, withscores=True)
            retry_after = int((oldest[0][1] + self.window - now) * 1000) if oldest else self.window * 1000
            return RateLimitResult(False, 0, 0, max(retry_after, 0))

        if current_tpm + estimated_tokens > tpm_limit:
            oldest = await self.client.zrange(tpm_key, 0, 0, withscores=True)
            retry_after = int((oldest[0][1] + self.window - now) * 1000) if oldest else self.window * 1000
            return RateLimitResult(False, rpm_limit - current_rpm, 0, max(retry_after, 0))

        # Increment: add current request with timestamp as score
        async with self.client.pipeline(transaction=True) as pipe:
            pipe.zadd(rpm_key, {f"{now}:{secrets.token_hex(4)}": now})
            pipe.zadd(tpm_key, {f"{now}:{secrets.token_hex(4)}": now})
            pipe.expire(rpm_key, self.window + 1)
            pipe.expire(tpm_key, self.window + 1)
            await pipe.execute()

        return RateLimitResult(
            True,
            rpm_limit - current_rpm - 1,
            tpm_limit - current_tpm - estimated_tokens
        )

    async def record_actual_tokens(self, scope_key: str, actual_tokens: int, rpm_limit: int, tpm_limit: int) -> None:
        """Call after upstream response to true-up token count."""
        now = time.time()
        tpm_key = f"rl:tpm:{scope_key}"
        # Replace the estimated entry with actual token weight
        # Simplified: add a correction entry; a production version would update the member score
        await self.client.zadd(tpm_key, {f"corr:{now}:{secrets.token_hex(4)}": now})
        await self.client.expire(tpm_key, self.window + 1)

Wire it into your request middleware:

# middleware.py
from fastapi import Request, HTTPException, Depends
from typing import Annotated

async def verify_api_key(
    request: Request,
    key_store: KeyStore = Depends(get_key_store),
    rate_limiter: SlidingWindowRateLimiter = Depends(get_rate_limiter)
) -> dict:
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        raise HTTPException(401, "Missing or invalid Authorization header")
    key = auth[7:]

    key_data = await key_store.lookup(key)
    if not key_data:
        raise HTTPException(401, "Invalid or revoked API key")

    scope_key = f"{key_data['scope_type']}:{key_data['scope_id']}"
    estimated_tokens = estimate_tokens_from_request(await request.body())

    rl_result = await rate_limiter.check_and_increment(
        scope_key,
        key_data["rate_limit_rpm"],
        key_data["rate_limit_tpm"],
        estimated_tokens
    )

    if not rl_result.allowed:
        raise HTTPException(
            429,
            "Rate limit exceeded",
            headers={"Retry-After": str((rl_result.retry_after_ms or 60000) // 1000)}
        )

    # Attach for downstream use
    request.state.key_data = key_data
    request.state.rate_limit = rl_result
    return key_data

Verify: Send 61 requests in 60 seconds with a 60 RPM key. Request 61 returns 429 with Retry-After. Confirm Redis sorted sets contain exactly 60 members after the burst.

Step 4: Enforce IP allowlists

Restrict keys to known infrastructure. Store CIDR blocks in ip_allowlist (empty = allow all). Check the client IP against the list before rate limiting.

# allowlist.py
import ipaddress
from typing import List

def ip_allowed(client_ip: str, allowlist: List[str]) -> bool:
    if not allowlist:
        return True
    try:
        client = ipaddress.ip_address(client_ip)
    except ValueError:
        return False
    for cidr in allowlist:
        try:
            network = ipaddress.ip_network(cidr, strict=False)
            if client in network:
                return True
        except ValueError:
            continue
    return False

def get_client_ip(request: Request) -> str:
    # Respect X-Forwarded-For from trusted proxy only
    forwarded = request.headers.get("X-Forwarded-For")
    if forwarded:
        # Take the first IP (original client) — assumes trusted proxy strips untrusted headers
        return forwarded.split(",")[0].strip()
    return request.client.host if request.client else "0.0.0.0"

Update middleware:

# middleware.py (add after key lookup)
client_ip = get_client_ip(request)
if not ip_allowed(client_ip, key_data["ip_allowlist"]):
    raise HTTPException(403, "IP not allowed for this API key")

Verify: Configure a key with ip_allowlist = ["10.0.0.0/8", "192.168.1.5/32"]. Request from 10.1.2.3 succeeds. Request from 203.0.113.10 returns 403. Confirm no rate limit counters incremented for rejected requests.

Step 5: Enforce domain allowlists

For browser-based or webhook callers, restrict the Origin or Referer header to approved domains. This mitigates key leakage via compromised frontends.

# allowlist.py (add)
from urllib.parse import urlparse

def domain_allowed(origin: str, allowlist: List[str]) -> bool:
    if not allowlist:
        return True
    if not origin:
        return False
    try:
        parsed = urlparse(origin)
        hostname = parsed.hostname or ""
    except Exception:
        return False
    # Allow exact match or subdomain match (e.g., allowlist "example.com" permits "app.example.com")
    for allowed in allowlist:
        if hostname == allowed or hostname.endswith("." + allowed):
            return True
    return False

Middleware addition:

origin = request.headers.get("Origin") or request.headers.get("Referer")
if not domain_allowed(origin, key_data["domain_allowlist"]):
    raise HTTPException(403, "Origin not allowed for this API key")

Verify: Key with domain_allowlist = ["app.example.com"]. Request with Origin: https://app.example.com passes. Origin: https://evil.com fails. Origin: https://admin.app.example.com passes (subdomain match). Missing Origin header fails.

Step 6: Enforce model allowlists

Prevent a key scoped for embeddings from calling expensive reasoning models. Store allowed model IDs in model_allowlist (empty = all models). Check against the model parameter in the request body.

# allowlist.py (add)
def model_allowed(requested_model: str, allowlist: List[str]) -> bool:
    if not allowlist:
        return True
    # Support wildcard prefixes: "gpt-4*" matches "gpt-4o", "gpt-4-turbo"
    for allowed in allowlist:
        if allowed.endswith("*"):
            if requested_model.startswith(allowed[:-1]):
                return True
        elif requested_model == allowed:
            return True
    return False

Middleware addition (parse JSON body once):

body = await request.json()
requested_model = body.get("model")
if requested_model and not model_allowed(requested_model, key_data["model_allowlist"]):
    raise HTTPException(403, f"Model '{requested_model}' not allowed for this API key")

Verify: Key with model_allowlist = ["gpt-4o*", "text-embedding-3-small"]. Request for gpt-4o-mini passes. Request for claude-3.5-sonnet fails. Request for gpt-4.1 fails. Empty allowlist permits any model.

Step 7: Implement key rotation and revocation

Rotation: issue a new key, mark old key revoked, keep both active for a grace period. Revocation: immediate disable by scope or individual key.

# key_lifecycle.py
from datetime import timedelta

class KeyManager:
    def __init__(self, store: KeyStore):
        self.store = store

    async def rotate_key(self, old_key: str, grace_period: timedelta = timedelta(hours=24)) -> str:
        key_data = await self.store.lookup(old_key)
        if not key_data:
            raise ValueError("Key not found")

        scope = KeyScope(ScopeType(key_data["scope_type"]), key_data["scope_id"], key_data["metadata"])
        policy = {
            "rpm": key_data["rate_limit_rpm"],
            "tpm": key_data["rate_limit_tpm"],
            "ip_allowlist": key_data["ip_allowlist"],
            "domain_allowlist": key_data["domain_allowlist"],
            "model_allowlist": key_data["model_allowlist"],
        }
        new_key = generate_key(scope)
        await self.store.create_key(new_key, scope, policy)

        # Schedule revocation of old key
        expires_at = datetime.now(timezone.utc) + grace_period
        async with self.store.pool.acquire() as conn:
            await conn.execute("""
                UPDATE api_keys SET expires_at = $1 WHERE key_hash = $2
            """, expires_at, self.store.hash_key(old_key))

        return new_key

    async def revoke_key(self, key: str) -> bool:
        key_hash = self.store.hash_key(key)
        async with self.store.pool.acquire() as conn:
            result = await conn.execute("""
                UPDATE api_keys SET status = 'revoked', revoked_at = now()
                WHERE key_hash = $1 AND status = 'active'
            """, key_hash)
            return result == "UPDATE 1"

    async def revoke_scope(self, scope_type: ScopeType, scope_id: str) -> int:
        return await self.store.revoke_by_scope(scope_type, scope_id)

Verify: Rotate a key. Confirm both old and new keys work during grace period. After expires_at, old key returns 401. Revoke a scope — all keys with that scope return 401 immediately.

Step 8: Add observability

Log every auth decision with structured fields. Emit metrics for dashboards and alerts.

# observability.py
import logging
from prometheus_client import Counter, Histogram, Gauge

auth_attempts = Counter("api_key_auth_attempts_total", "Total auth attempts", ["result", "scope_type"])
rate_limit_hits = Counter("api_key_rate_limit_hits_total", "Rate limit rejections", ["scope_type", "limit_type"])
request_latency = Histogram("api_key_auth_latency_seconds", "Auth middleware latency")
active_keys = Gauge("api_key_active_total", "Active keys by scope", ["scope_type", "scope_id"])

logger = logging.getLogger("api_keys")

def log_auth_decision(key_data: dict, allowed: bool, reason: str = "", **extra):
    fields = {
        "scope_type": key_data["scope_type"],
        "scope_id": key_data["scope_id"],
        "key_prefix": key_data["key_prefix"],
        "allowed": allowed,
        "reason": reason,
        **extra
    }
    if allowed:
        logger.info("API key authorized", extra=fields)
        auth_attempts.labels(result="allowed", scope_type=key_data["scope_type"]).inc()
    else:
        logger.warning("API key denied", extra=fields)
        auth_attempts.labels(result="denied", scope_type=key_data["scope_type"]).inc()

def record_rate_limit_hit(scope_type: str, limit_type: str):
    rate_limit_hits.labels(scope_type=scope_type, limit_type=limit_type).inc()

def update_active_keys_gauge(pool: asyncpg.Pool):
    """Run periodically to refresh gauge."""
    async def _refresh():
        async with pool.acquire() as conn:
            rows = await conn.fetch("""
                SELECT scope_type, scope_id, COUNT(*) as cnt
                FROM api_keys WHERE status = 'active'
                GROUP BY scope_type, scope_id
            """)
            for row in rows:
                active_keys.labels(scope_type=row["scope_type"], scope_id=row["scope_id"]).set(row["cnt"])
    return _refresh

Middleware integration:

# middleware.py (wrap verify_api_key)
@request_latency.time()
async def verify_api_key(...):
    start = time.time()
    try:
        key_data = await _verify_impl(...)
        log_auth_decision(key_data, True)
        return key_data
    except HTTPException as e:
        log_auth_decision(
            getattr(request.state, "key_data", {}),
            False,
            reason=e.detail,
            status_code=e.status_code
        )
        if e.status_code == 429:
            record_rate_limit_hit(key_data.get("scope_type", "unknown"), "rpm" if "minute" in e.detail else "tpm")
        raise

Verify: Generate traffic. Confirm api_key_auth_attempts_total increments for allowed/denied. Confirm api_key_rate_limit_hits_total increments on 429. Grafana dashboard shows active keys per tenant. Alert on rate(rate_limit_hits[5m]) > 0.1 for any scope.

Step 9: Test end-to-end with a contract test suite

Write a pytest suite that exercises the full chain: key creation, valid request, rate limit, IP block, domain block, model block, rotation, revocation.

# test_api_keys.py
import pytest
from httpx import AsyncClient, ASGITransport
from main import app  # your FastAPI app

@pytest.fixture
async def client():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c

@pytest.fixture
def tenant_key(key_manager):
    scope = KeyScope(ScopeType.TENANT, "test_corp", {"tier": "pro"})
    policy = {"rpm": 10, "tpm": 1000, "ip_allowlist": ["10.0.0.0/8"], "model_allowlist": ["gpt-4o*"]}
    key = generate_key(scope)
    await key_manager.store.create_key(key, scope, policy)
    return key

async def test_valid_request(client, tenant_key):
    resp = await client.post("/v1/chat/completions",
        headers={"Authorization": f"Bearer {tenant_key}"},
        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]})
    assert resp.status_code == 200

async def test_rate_limit_rpm(client, tenant_key):
    for i in range(10):
        resp = await client.post("/v1/chat/completions",
            headers={"Authorization": f"Bearer {tenant_key}"},
            json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": f"msg {i}"}]})
        assert resp.status_code == 200
    # 11th request should 429
    resp = await client.post("/v1/chat/completions",
        headers={"Authorization": f"Bearer {tenant_key}"},
        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "overflow"}]})
    assert resp.status_code == 429
    assert "Retry-After" in resp.headers

async def test_ip_allowlist(client, tenant_key):
    # Simulate request from non-allowed IP via test client header override
    resp = await client.post("/v1/chat/completions",
        headers={"Authorization": f"Bearer {tenant_key}", "X-Forwarded-For": "203.0.113.10"},
        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]})
    assert resp.status_code == 403

async def test_model_allowlist(client, tenant_key):
    resp = await client.post("/v1/chat/completions",
        headers={"Authorization": f"Bearer {tenant_key}"},
        json={"model": "claude-3.5-sonnet", "messages": [{"role": "user", "content": "hi"}]})
    assert resp.status_code == 403

async def test_key_rotation(key_manager, tenant_key):
    new_key = await key_manager.rotate_key(tenant_key)
    # Old key still works during grace period
    resp = await client.post("/v1/chat/completions",
        headers={"Authorization": f"Bearer {tenant_key}"},
        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]})
    assert resp.status_code == 200
    # New key works
    resp = await client.post("/v1/chat/completions",
        headers={"Authorization": f"Bearer {new_key}"},
        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]})
    assert resp.status_code == 200

async def test_scope_revocation(key_manager, client, tenant_key):
    await key_manager.revoke_scope(ScopeType.TENANT, "test_corp")
    resp = await client.post("/v1/chat/completions",
        headers={"Authorization": f"Bearer {tenant_key}"},
        json={"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]})
    assert resp.status_code == 401

Run in CI on every deploy. Verify: All tests pass. Mutation testing (flip an allowlist entry) catches regressions.

Step 10: Operational playbook

Document runbooks for common incidents:

Keep a CLI tool for ops:

# ops/keyctl.py
import asyncio, click
from key_store import KeyStore
from key_lifecycle import KeyManager
from scopes import KeyScope, ScopeType

@click.group()
def cli(): pass

@cli.command()
@click.argument("scope_type", type=click.Choice([t.value for t in ScopeType]))
@click.argument("scope_id")
@click.option("--rpm", default=60)
@click.option("--tpm", default=100000)
@click.option("--ip", "ip_allowlist", multiple=True)
@click.option("--domain", "domain_allowlist", multiple=True)
@click.option("--model", "model_allowlist", multiple=True)
def create(scope_type, scope_id, rpm, tpm, ip_allowlist, domain_allowlist, model_allowlist):
    async def _create():
        pool = await asyncpg.create_pool(DATABASE_URL)
        store = KeyStore(pool)
        mgr = KeyManager(store)
        scope = KeyScope(ScopeType(scope_type), scope_id, {})
        policy = {"rpm": rpm, "tpm": tpm,
                  "ip_allowlist": list(ip_allowlist),
                  "domain_allowlist": list(domain_allowlist),
                  "model_allowlist": list(model_allowlist)}
        key = generate_key(scope)
        await store.create_key(key, scope, policy)
        print(key)
    asyncio.run(_create())

@cli.command()
@click.argument("key")
def revoke(key):
    async def _revoke():
        pool = await asyncpg.create_pool(DATABASE_URL)
        store = KeyStore(pool)
        mgr = KeyManager(store)
        ok = await mgr.revoke_key(key)
        print("Revoked" if ok else "Not found")
    asyncio.run(_revoke())

Verify: Run keyctl create tenant acme_corp --rpm 120 --model "gpt-4o*". Key appears in DB. keyctl revoke <key> marks it revoked. Subsequent requests return 401.

Scoping API keys with rate limits and allowlists is not optional for production LLM workloads. The pattern above — prefix-encoded scopes, Redis sliding windows, CIDR/domain/model allowlists, rotation with grace periods, and full observability — gives you the control plane to run multi-tenant inference without surprises. Start with the schema and middleware, add the contract tests, then layer on the runbooks. Your future on-call self will thank you.