Implementing LLM Gateway Middleware: Centralized Auth, Rate Limiting, Audit Logging, and Fallback Routing Across Multiple Providers

Introduction

Your team just shipped a product that calls GPT-4o for document summarization, Claude for customer support, and Gemini for embeddings. Three different API keys floating across three different services, each dev managing their own retry logic, each microservice implementing authentication differently, and no unified view of how many tokens you burned this month or which requests failed. Then the OpenAI API goes down during a product demo, and there's no fallback — just a 500 error and an awkward silence.

This is the problem an LLM gateway solves. A gateway sits between your application layer and the upstream LLM providers, acting as a single choke point through which every AI request flows. That choke point lets you centralize authentication, enforce rate limits per tenant or user, log every request for compliance and debugging, and reroute traffic when a provider fails. Done right, it transforms a fragile collection of direct API calls into a resilient, observable, governable infrastructure layer. Done wrong, it becomes a single point of failure with a false sense of security.

By the end of this lesson, you'll have built a production-grade LLM gateway middleware in Python, walked through every major design decision, and understood the trade-offs at each step. This isn't a wrapper around an existing tool — we're building from first principles so you understand the internals, which makes you equally capable of evaluating, extending, or debugging any existing gateway solution.

What you'll learn:

• How to design a provider-agnostic request/response abstraction layer that normalizes differences between OpenAI, Anthropic, and Google APIs
• How to implement centralized API key management and per-tenant authentication with JWT-based identity propagation
• How to build multi-tier rate limiting (per user, per tenant, per provider) using token bucket algorithms backed by Redis
• How to construct a structured audit logging pipeline that captures latency, token usage, costs, and failure modes without blocking the critical path
• How to implement priority-ordered fallback routing with circuit breakers that detect provider degradation and route around it automatically

---

Prerequisites

You should be comfortable with Python async programming (asyncio, aiohttp), have a working knowledge of Redis, understand basic HTTP middleware patterns, and have used at least one LLM provider API directly. Familiarity with JWT tokens and Docker is helpful but not required. You'll need Python 3.11+, Redis 7+, and API keys for at least two LLM providers to follow along.

---

Designing the Gateway Architecture Before Writing a Line of Code

The biggest mistake teams make when building a gateway is starting with code instead of starting with a data model. Before you write a single function, you need to answer three questions: What is the canonical request format? What is the canonical response format? And what state does the gateway need to maintain between requests?

The canonical request model is the single hardest design decision in the entire system. Every LLM provider has a different schema. OpenAI uses messages with role and content. Anthropic uses messages with a system parameter at the top level and a maximum token parameter called max_tokens. Google's Gemini uses contents with parts. If you let these differences leak into your gateway, you'll spend the rest of your life writing provider-specific code paths everywhere.

The right approach is to define your own internal schema that is strictly richer than any single provider's schema, then write translators at the edges — thin adapters that convert from your canonical format into whatever the provider expects. This is the Adapter pattern, and it's the only way to keep the gateway core clean.

Here's the canonical data model we'll build the rest of the system around:

# gateway/models.py
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Optional
import uuid
import time


class Provider(str, Enum):
    OPENAI = "openai"
    ANTHROPIC = "anthropic"
    GEMINI = "gemini"


class MessageRole(str, Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"


@dataclass
class Message:
    role: MessageRole
    content: str


@dataclass
class GatewayRequest:
    messages: list[Message]
    model: str                          # logical model name, e.g. "gpt-4o"
    max_tokens: int = 1024
    temperature: float = 0.7
    tenant_id: str = ""
    user_id: str = ""
    request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: float = field(default_factory=time.time)
    metadata: dict[str, Any] = field(default_factory=dict)
    # Which providers to try, in order. Empty means use routing table.
    provider_preference: list[Provider] = field(default_factory=list)


@dataclass
class TokenUsage:
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0


@dataclass
class GatewayResponse:
    content: str
    model: str
    provider: Provider
    request_id: str
    usage: TokenUsage
    latency_ms: float
    cost_usd: float
    cached: bool = False
    fallback_used: bool = False
    attempts: int = 1

Notice what GatewayRequest includes beyond a raw API payload: tenant_id, user_id, request_id, and metadata. These are the fields that make the gateway useful. Without tenant identity, you can't do per-tenant rate limiting or cost attribution. Without a stable request_id, your audit logs are useless for debugging. Without metadata, you can't attach business context (like {"feature": "document_summary", "document_id": "doc_8821"}) that makes logs searchable.

The provider_preference list on the request is interesting. Most of the time it'll be empty, and the gateway's routing table decides which provider to use based on the model name. But sometimes callers need to express explicit preferences — for example, a compliance requirement that customer data only goes to Anthropic, not OpenAI.

---

Building the Provider Adapter Layer

Each provider adapter is responsible for exactly one thing: translating between the canonical format and the provider's wire format. No business logic. No retry handling. No rate limiting. Just translation and the raw HTTP call.

# gateway/adapters/base.py
from abc import ABC, abstractmethod
from gateway.models import GatewayRequest, GatewayResponse
import asyncio


class ProviderAdapter(ABC):
    """Base class for all provider adapters."""

    @abstractmethod
    async def complete(self, request: GatewayRequest) -> GatewayResponse:
        """Execute a completion request against this provider."""
        ...

    @abstractmethod
    def supports_model(self, model: str) -> bool:
        """Returns True if this adapter can handle the requested model."""
        ...

    @abstractmethod
    def estimate_cost(self, usage: dict, model: str) -> float:
        """Estimate cost in USD given token usage and model."""
        ...

Now let's implement the OpenAI adapter. Pay attention to how it handles the timing and error classification:

# gateway/adapters/openai_adapter.py
import time
import aiohttp
from gateway.adapters.base import ProviderAdapter
from gateway.models import (
    GatewayRequest, GatewayResponse, Provider, TokenUsage, MessageRole
)


OPENAI_COST_PER_1K = {
    "gpt-4o": {"prompt": 0.005, "completion": 0.015},
    "gpt-4o-mini": {"prompt": 0.00015, "completion": 0.0006},
    "gpt-3.5-turbo": {"prompt": 0.0005, "completion": 0.0015},
}


class OpenAIAdapter(ProviderAdapter):
    def __init__(self, api_key: str, base_url: str = "https://api.openai.com/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self._session: aiohttp.ClientSession | None = None

    async def _get_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            timeout = aiohttp.ClientTimeout(total=120, connect=5)
            self._session = aiohttp.ClientSession(timeout=timeout)
        return self._session

    def supports_model(self, model: str) -> bool:
        return model in OPENAI_COST_PER_1K

    def estimate_cost(self, usage: dict, model: str) -> float:
        rates = OPENAI_COST_PER_1K.get(model, {"prompt": 0.01, "completion": 0.03})
        prompt_cost = (usage.get("prompt_tokens", 0) / 1000) * rates["prompt"]
        completion_cost = (usage.get("completion_tokens", 0) / 1000) * rates["completion"]
        return prompt_cost + completion_cost

    def _build_payload(self, request: GatewayRequest) -> dict:
        messages = []
        for msg in request.messages:
            messages.append({"role": msg.role.value, "content": msg.content})
        return {
            "model": request.model,
            "messages": messages,
            "max_tokens": request.max_tokens,
            "temperature": request.temperature,
        }

    async def complete(self, request: GatewayRequest) -> GatewayResponse:
        session = await self._get_session()
        payload = self._build_payload(request)
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

        start_time = time.monotonic()

        async with session.post(
            f"{self.base_url}/chat/completions",
            json=payload,
            headers=headers,
        ) as resp:
            elapsed_ms = (time.monotonic() - start_time) * 1000

            if resp.status == 429:
                raise RateLimitError(f"OpenAI rate limit: {await resp.text()}")
            if resp.status == 503:
                raise ProviderUnavailableError(f"OpenAI unavailable: {await resp.text()}")
            if resp.status >= 400:
                raise ProviderError(f"OpenAI error {resp.status}: {await resp.text()}")

            data = await resp.json()

        usage_raw = data.get("usage", {})
        usage = TokenUsage(
            prompt_tokens=usage_raw.get("prompt_tokens", 0),
            completion_tokens=usage_raw.get("completion_tokens", 0),
            total_tokens=usage_raw.get("total_tokens", 0),
        )

        return GatewayResponse(
            content=data["choices"][0]["message"]["content"],
            model=request.model,
            provider=Provider.OPENAI,
            request_id=request.request_id,
            usage=usage,
            latency_ms=elapsed_ms,
            cost_usd=self.estimate_cost(usage_raw, request.model),
        )


class RateLimitError(Exception):
    """Provider returned 429."""
    pass

class ProviderUnavailableError(Exception):
    """Provider returned 5xx or is unreachable."""
    pass

class ProviderError(Exception):
    """Generic provider error."""
    pass

The Anthropic adapter follows the same structure but handles the API's different shape:

# gateway/adapters/anthropic_adapter.py
import time
import aiohttp
from gateway.adapters.base import ProviderAdapter
from gateway.adapters.openai_adapter import RateLimitError, ProviderUnavailableError, ProviderError
from gateway.models import GatewayRequest, GatewayResponse, Provider, TokenUsage, MessageRole


ANTHROPIC_COST_PER_1K = {
    "claude-3-5-sonnet-20241022": {"prompt": 0.003, "completion": 0.015},
    "claude-3-haiku-20240307": {"prompt": 0.00025, "completion": 0.00125},
}


class AnthropicAdapter(ProviderAdapter):
    def __init__(self, api_key: str):
        self.api_key = api_key
        self._session: aiohttp.ClientSession | None = None

    async def _get_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            timeout = aiohttp.ClientTimeout(total=120, connect=5)
            self._session = aiohttp.ClientSession(timeout=timeout)
        return self._session

    def supports_model(self, model: str) -> bool:
        return model in ANTHROPIC_COST_PER_1K

    def estimate_cost(self, usage: dict, model: str) -> float:
        rates = ANTHROPIC_COST_PER_1K.get(model, {"prompt": 0.003, "completion": 0.015})
        return (
            (usage.get("input_tokens", 0) / 1000) * rates["prompt"] +
            (usage.get("output_tokens", 0) / 1000) * rates["completion"]
        )

    def _build_payload(self, request: GatewayRequest) -> tuple[dict, str]:
        """Returns (payload, system_prompt)"""
        system = ""
        messages = []
        for msg in request.messages:
            if msg.role == MessageRole.SYSTEM:
                system = msg.content  # Anthropic takes system at top level
            else:
                messages.append({"role": msg.role.value, "content": msg.content})
        payload = {
            "model": request.model,
            "messages": messages,
            "max_tokens": request.max_tokens,
            "temperature": request.temperature,
        }
        if system:
            payload["system"] = system
        return payload

    async def complete(self, request: GatewayRequest) -> GatewayResponse:
        session = await self._get_session()
        payload = self._build_payload(request)
        headers = {
            "x-api-key": self.api_key,
            "anthropic-version": "2023-06-01",
            "Content-Type": "application/json",
        }

        start_time = time.monotonic()
        async with session.post(
            "https://api.anthropic.com/v1/messages",
            json=payload,
            headers=headers,
        ) as resp:
            elapsed_ms = (time.monotonic() - start_time) * 1000
            if resp.status == 429:
                raise RateLimitError(f"Anthropic rate limit")
            if resp.status >= 500:
                raise ProviderUnavailableError(f"Anthropic unavailable: {resp.status}")
            if resp.status >= 400:
                raise ProviderError(f"Anthropic error {resp.status}: {await resp.text()}")
            data = await resp.json()

        usage_raw = data.get("usage", {})
        usage = TokenUsage(
            prompt_tokens=usage_raw.get("input_tokens", 0),
            completion_tokens=usage_raw.get("output_tokens", 0),
            total_tokens=usage_raw.get("input_tokens", 0) + usage_raw.get("output_tokens", 0),
        )

        return GatewayResponse(
            content=data["content"][0]["text"],
            model=request.model,
            provider=Provider.ANTHROPIC,
            request_id=request.request_id,
            usage=usage,
            latency_ms=elapsed_ms,
            cost_usd=self.estimate_cost(usage_raw, request.model),
        )

The key insight here: The adapters are the only place where provider-specific knowledge lives. Every other component in the gateway — rate limiter, logger, circuit breaker — works entirely with the canonical GatewayRequest and GatewayResponse types. This means you can add a new provider by adding one file with one class, with zero changes to the rest of the system.

---

Centralized Authentication and Tenant Identity

The gateway is a natural enforcement point for authentication, but the design requires careful thought. There are two distinct authentication concerns: authenticating the caller to the gateway, and authenticating the gateway to the provider. These must never be conflated.

Callers authenticate to the gateway using JWT tokens. Each JWT contains the tenant ID, user ID, and a set of allowed models/providers. The gateway verifies the JWT, extracts identity, and attaches it to the canonical request. Provider credentials live only in the gateway's configuration — callers never see them.

# gateway/auth.py
import jwt
import time
from dataclasses import dataclass
from typing import Optional


@dataclass
class CallerIdentity:
    tenant_id: str
    user_id: str
    allowed_models: list[str]   # empty means all models allowed
    rate_limit_tier: str        # "free", "standard", "enterprise"
    token_budget_daily: int     # max tokens per day, -1 means unlimited


class AuthError(Exception):
    pass


class GatewayAuthMiddleware:
    def __init__(self, jwt_secret: str, algorithm: str = "HS256"):
        self.jwt_secret = jwt_secret
        self.algorithm = algorithm

    def authenticate(self, token: str) -> CallerIdentity:
        """Verify JWT and return caller identity. Raises AuthError on failure."""
        try:
            payload = jwt.decode(
                token,
                self.jwt_secret,
                algorithms=[self.algorithm],
                options={"require": ["exp", "sub", "tenant_id"]},
            )
        except jwt.ExpiredSignatureError:
            raise AuthError("Token has expired")
        except jwt.InvalidTokenError as e:
            raise AuthError(f"Invalid token: {e}")

        if payload["exp"] < time.time():
            raise AuthError("Token expired")

        return CallerIdentity(
            tenant_id=payload["tenant_id"],
            user_id=payload["sub"],
            allowed_models=payload.get("allowed_models", []),
            rate_limit_tier=payload.get("rate_limit_tier", "standard"),
            token_budget_daily=payload.get("token_budget_daily", -1),
        )

    def authorize_model(self, identity: CallerIdentity, model: str) -> None:
        """Check if the caller is allowed to use this model."""
        if identity.allowed_models and model not in identity.allowed_models:
            raise AuthError(
                f"Tenant '{identity.tenant_id}' is not authorized to use model '{model}'"
            )

Provider credentials live in a CredentialStore that is completely separate from auth:

# gateway/credentials.py
import os
from gateway.models import Provider


class CredentialStore:
    """
    Manages provider API keys. In production, back this with
    AWS Secrets Manager, HashiCorp Vault, or similar.
    For now, environment variables with a clear naming convention.
    """
    def __init__(self):
        self._keys: dict[Provider, str] = {}
        self._load_from_env()

    def _load_from_env(self):
        mapping = {
            Provider.OPENAI: "GATEWAY_OPENAI_API_KEY",
            Provider.ANTHROPIC: "GATEWAY_ANTHROPIC_API_KEY",
            Provider.GEMINI: "GATEWAY_GEMINI_API_KEY",
        }
        for provider, env_var in mapping.items():
            key = os.getenv(env_var)
            if key:
                self._keys[provider] = key

    def get_key(self, provider: Provider) -> str:
        key = self._keys.get(provider)
        if not key:
            raise ValueError(f"No API key configured for provider: {provider}")
        return key

    def available_providers(self) -> list[Provider]:
        return list(self._keys.keys())

Security note: Never log provider API keys, never include them in responses, and never accept them from callers. The credential store should be initialized once at startup and read from a secrets management system — not .env files in production. The environment variable approach here is for local development only.

---

Multi-Tier Rate Limiting with Token Buckets

Rate limiting is where most gateway implementations go wrong. They implement a simple request-per-minute counter when what you actually need is a multi-dimensional rate limiter that tracks: requests per user per minute, tokens per tenant per day, and requests per provider per minute (to stay within your own provider quotas).

We'll use the token bucket algorithm because it allows bursting, which is the right behavior for an API gateway. A fixed window counter creates artificial cliffs — a user who makes 60 requests at 12:00:59 and 60 more at 12:01:01 has made 120 requests in two seconds but violated no per-minute limit. Token bucket handles this correctly by refilling at a constant rate.

Redis is the right storage backend here because it's atomic, fast, and shared across all gateway instances (which you'll inevitably run more than one of).

# gateway/rate_limiter.py
import time
import redis.asyncio as aioredis
from dataclasses import dataclass
from gateway.models import GatewayRequest, Provider


@dataclass
class RateLimitConfig:
    # Requests per minute per user
    user_rpm: int = 20
    # Requests per minute per tenant
    tenant_rpm: int = 200
    # Tokens per day per tenant
    tenant_tpd: int = 5_000_000
    # Requests per minute to each provider (to protect your own quotas)
    provider_rpm: dict[str, int] = None

    def __post_init__(self):
        if self.provider_rpm is None:
            self.provider_rpm = {
                Provider.OPENAI: 500,
                Provider.ANTHROPIC: 200,
                Provider.GEMINI: 300,
            }


TIER_CONFIGS = {
    "free": RateLimitConfig(user_rpm=5, tenant_rpm=50, tenant_tpd=100_000),
    "standard": RateLimitConfig(user_rpm=20, tenant_rpm=200, tenant_tpd=5_000_000),
    "enterprise": RateLimitConfig(user_rpm=100, tenant_rpm=2000, tenant_tpd=50_000_000),
}


class RateLimitExceeded(Exception):
    def __init__(self, message: str, retry_after_seconds: float):
        super().__init__(message)
        self.retry_after_seconds = retry_after_seconds


class RedisTokenBucketLimiter:
    """
    Token bucket rate limiter backed by Redis.
    Uses a Lua script for atomic check-and-consume to avoid race conditions.
    """

    # Lua script: atomically check capacity and consume tokens
    # Returns [allowed (1/0), current_tokens, ttl_ms]
    CONSUME_SCRIPT = """
    local key = KEYS[1]
    local capacity = tonumber(ARGV[1])
    local refill_rate = tonumber(ARGV[2])  -- tokens per second
    local now = tonumber(ARGV[3])          -- current time in ms
    local requested = tonumber(ARGV[4])

    local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
    local tokens = tonumber(bucket[1]) or capacity
    local last_refill = tonumber(bucket[2]) or now

    -- Calculate how many tokens to add since last check
    local elapsed = math.max(0, now - last_refill)
    local new_tokens = math.min(capacity, tokens + (elapsed / 1000) * refill_rate)

    if new_tokens >= requested then
        new_tokens = new_tokens - requested
        redis.call('HMSET', key, 'tokens', new_tokens, 'last_refill', now)
        redis.call('EXPIRE', key, 3600)
        return {1, math.floor(new_tokens), 0}
    else
        local wait_ms = math.ceil((requested - new_tokens) / refill_rate * 1000)
        redis.call('HMSET', key, 'tokens', new_tokens, 'last_refill', now)
        redis.call('EXPIRE', key, 3600)
        return {0, math.floor(new_tokens), wait_ms}
    end
    """

    def __init__(self, redis_client: aioredis.Redis):
        self.redis = redis_client
        self._script = None

    async def _get_script(self):
        if self._script is None:
            self._script = self.redis.register_script(self.CONSUME_SCRIPT)
        return self._script

    async def check_and_consume(
        self,
        key: str,
        capacity: int,
        refill_rate: float,
        requested: int = 1,
    ) -> tuple[bool, float]:
        """
        Returns (allowed, retry_after_seconds).
        retry_after_seconds is 0 if allowed.
        """
        script = await self._get_script()
        now_ms = int(time.time() * 1000)
        result = await script(
            keys=[key],
            args=[capacity, refill_rate, now_ms, requested],
        )
        allowed = bool(result[0])
        retry_after = result[2] / 1000.0  # convert ms to seconds
        return allowed, retry_after

    async def check_request(
        self,
        request: GatewayRequest,
        tier: str,
        provider: Provider,
    ) -> None:
        """
        Run all rate limit checks for a request.
        Raises RateLimitExceeded with retry-after info if any check fails.
        """
        config = TIER_CONFIGS.get(tier, TIER_CONFIGS["standard"])

        checks = [
            (
                f"rl:user:{request.tenant_id}:{request.user_id}:rpm",
                config.user_rpm,
                config.user_rpm / 60.0,
                "User rate limit exceeded",
            ),
            (
                f"rl:tenant:{request.tenant_id}:rpm",
                config.tenant_rpm,
                config.tenant_rpm / 60.0,
                "Tenant rate limit exceeded",
            ),
            (
                f"rl:provider:{provider.value}:rpm",
                config.provider_rpm[provider],
                config.provider_rpm[provider] / 60.0,
                f"Provider {provider.value} rate limit exceeded",
            ),
        ]

        for key, capacity, refill_rate, message in checks:
            allowed, retry_after = await self.check_and_consume(
                key, capacity, refill_rate
            )
            if not allowed:
                raise RateLimitExceeded(message, retry_after)

    async def record_token_usage(
        self,
        tenant_id: str,
        tokens_used: int,
        tier: str,
    ) -> None:
        """
        Track daily token consumption. Call this after a successful completion.
        Uses a 24-hour expiring counter.
        """
        config = TIER_CONFIGS.get(tier, TIER_CONFIGS["standard"])
        key = f"tokens:{tenant_id}:{self._today_key()}"
        
        pipe = self.redis.pipeline()
        pipe.incrby(key, tokens_used)
        pipe.expire(key, 86400)
        results = await pipe.execute()
        
        total = results[0]
        if config.tenant_tpd > 0 and total > config.tenant_tpd:
            raise RateLimitExceeded(
                f"Daily token budget exceeded for tenant {tenant_id}",
                retry_after_seconds=self._seconds_until_midnight(),
            )

    def _today_key(self) -> str:
        from datetime import date
        return date.today().isoformat()

    def _seconds_until_midnight(self) -> float:
        from datetime import datetime, time as dtime
        now = datetime.now()
        midnight = datetime.combine(now.date(), dtime.max)
        return (midnight - now).total_seconds()

Why Lua scripts for the bucket? Redis is single-threaded, but a read-modify-write sequence across multiple commands isn't atomic — another client could slip in between your GET and SET. The Lua script executes atomically on the Redis server, eliminating this race condition entirely. This matters when you're running multiple gateway instances.

---

Circuit Breakers and Fallback Routing

Rate limiting protects your providers from you. Circuit breakers protect you from your providers. When a provider starts returning errors at high rates, you want to stop sending requests to it immediately — not after timing out on 100 requests in a row.

The circuit breaker has three states. Closed is normal operation: requests flow through. Open means the provider is considered down: requests immediately fail or route to a fallback without even attempting the provider. Half-open is the recovery state: after a timeout, we allow a single test request through. If it succeeds, we close the circuit. If it fails, we reopen it.

# gateway/circuit_breaker.py
import time
import asyncio
from enum import Enum
from dataclasses import dataclass, field
from gateway.models import Provider


class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5          # failures before opening
    success_threshold: int = 2          # successes in half-open before closing
    timeout_seconds: float = 60.0       # time before trying again (open -> half-open)
    failure_window_seconds: float = 60.0 # rolling window for failure counting


@dataclass
class CircuitBreakerState:
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    success_count: int = 0
    last_failure_time: float = 0.0
    last_state_change: float = field(default_factory=time.time)
    failure_timestamps: list[float] = field(default_factory=list)


class CircuitBreakerOpen(Exception):
    def __init__(self, provider: Provider, retry_after: float):
        self.provider = provider
        self.retry_after = retry_after
        super().__init__(f"Circuit open for {provider.value}, retry after {retry_after:.1f}s")


class ProviderCircuitBreaker:
    def __init__(self, config: CircuitBreakerConfig = None):
        self.config = config or CircuitBreakerConfig()
        self._states: dict[Provider, CircuitBreakerState] = {
            p: CircuitBreakerState() for p in Provider
        }
        self._half_open_lock: dict[Provider, asyncio.Lock] = {
            p: asyncio.Lock() for p in Provider
        }

    def get_state(self, provider: Provider) -> CircuitState:
        state = self._states[provider]
        if state.state == CircuitState.OPEN:
            elapsed = time.time() - state.last_state_change
            if elapsed >= self.config.timeout_seconds:
                state.state = CircuitState.HALF_OPEN
                state.success_count = 0
                return CircuitState.HALF_OPEN
        return state.state

    def check(self, provider: Provider) -> None:
        """Raises CircuitBreakerOpen if provider circuit is open."""
        current_state = self.get_state(provider)
        if current_state == CircuitState.OPEN:
            state = self._states[provider]
            retry_after = self.config.timeout_seconds - (
                time.time() - state.last_state_change
            )
            raise CircuitBreakerOpen(provider, max(0, retry_after))

    def record_success(self, provider: Provider) -> None:
        state = self._states[provider]
        now = time.time()

        if state.state == CircuitState.HALF_OPEN:
            state.success_count += 1
            if state.success_count >= self.config.success_threshold:
                state.state = CircuitState.CLOSED
                state.failure_count = 0
                state.failure_timestamps.clear()
                state.last_state_change = now
        elif state.state == CircuitState.CLOSED:
            # Prune old failures from the window
            cutoff = now - self.config.failure_window_seconds
            state.failure_timestamps = [t for t in state.failure_timestamps if t > cutoff]
            state.failure_count = len(state.failure_timestamps)

    def record_failure(self, provider: Provider) -> None:
        state = self._states[provider]
        now = time.time()

        state.failure_timestamps.append(now)
        # Prune outside window
        cutoff = now - self.config.failure_window_seconds
        state.failure_timestamps = [t for t in state.failure_timestamps if t > cutoff]
        state.failure_count = len(state.failure_timestamps)
        state.last_failure_time = now

        if state.state in (CircuitState.CLOSED, CircuitState.HALF_OPEN):
            if state.failure_count >= self.config.failure_threshold:
                state.state = CircuitState.OPEN
                state.last_state_change = now

Now we build the routing layer that ties circuit breakers and adapters together. The routing table maps models to an ordered list of providers. When the primary fails, the gateway tries the next one:

# gateway/router.py
from gateway.models import GatewayRequest, GatewayResponse, Provider
from gateway.adapters.base import ProviderAdapter
from gateway.adapters.openai_adapter import RateLimitError, ProviderUnavailableError, ProviderError
from gateway.circuit_breaker import ProviderCircuitBreaker, CircuitBreakerOpen
import logging

logger = logging.getLogger(__name__)

# Model name -> ordered list of providers to try
DEFAULT_ROUTING_TABLE: dict[str, list[Provider]] = {
    "gpt-4o": [Provider.OPENAI],
    "gpt-4o-mini": [Provider.OPENAI],
    "claude-3-5-sonnet-20241022": [Provider.ANTHROPIC],
    "claude-3-haiku-20240307": [Provider.ANTHROPIC],
    # For models that have equivalents across providers,
    # define fallback chains
    "smart": [Provider.OPENAI, Provider.ANTHROPIC],     # logical alias
    "fast": [Provider.ANTHROPIC, Provider.OPENAI],
}

# When falling back, what model to use for each provider
FALLBACK_MODEL_MAP: dict[Provider, str] = {
    Provider.OPENAI: "gpt-4o",
    Provider.ANTHROPIC: "claude-3-5-sonnet-20241022",
}


class GatewayRouter:
    def __init__(
        self,
        adapters: dict[Provider, ProviderAdapter],
        circuit_breaker: ProviderCircuitBreaker,
        routing_table: dict[str, list[Provider]] = None,
    ):
        self.adapters = adapters
        self.circuit_breaker = circuit_breaker
        self.routing_table = routing_table or DEFAULT_ROUTING_TABLE

    def _get_provider_order(self, request: GatewayRequest) -> list[Provider]:
        """Determine which providers to try, in order."""
        if request.provider_preference:
            return request.provider_preference
        return self.routing_table.get(request.model, [Provider.OPENAI])

    async def route(self, request: GatewayRequest) -> GatewayResponse:
        providers = self._get_provider_order(request)
        last_exception = None
        attempts = 0

        for i, provider in enumerate(providers):
            attempts += 1
            is_fallback = i > 0

            try:
                self.circuit_breaker.check(provider)
            except CircuitBreakerOpen as e:
                logger.warning(
                    f"Circuit open for {provider.value}, skipping. "
                    f"request_id={request.request_id}"
                )
                last_exception = e
                continue

            # Adjust the model name if this is a fallback to a different provider
            routed_request = request
            if is_fallback and provider in FALLBACK_MODEL_MAP:
                routed_request = GatewayRequest(
                    **{**request.__dict__,
                       "model": FALLBACK_MODEL_MAP[provider]}
                )

            adapter = self.adapters.get(provider)
            if not adapter:
                logger.error(f"No adapter configured for {provider.value}")
                continue

            try:
                response = await adapter.complete(routed_request)
                self.circuit_breaker.record_success(provider)

                if is_fallback:
                    response.fallback_used = True
                response.attempts = attempts
                return response

            except RateLimitError as e:
                logger.warning(f"Rate limited by {provider.value}: {e}")
                # Don't record as circuit breaker failure — provider is fine, just busy
                last_exception = e
                continue

            except (ProviderUnavailableError, ProviderError) as e:
                logger.error(f"Provider {provider.value} failed: {e}")
                self.circuit_breaker.record_failure(provider)
                last_exception = e
                continue

            except Exception as e:
                logger.exception(f"Unexpected error from {provider.value}")
                self.circuit_breaker.record_failure(provider)
                last_exception = e
                continue

        raise Exception(
            f"All providers exhausted after {attempts} attempts. "
            f"Last error: {last_exception}"
        )

Subtle design decision: We differentiate between RateLimitError and ProviderUnavailableError in the circuit breaker logic. A 429 rate limit means the provider is healthy but we're asking too much. Recording it as a circuit breaker failure would be wrong — it would cause us to stop sending requests to a perfectly functional provider. Only genuine errors (5xx, timeouts, connection failures) should trip the circuit breaker.

---

Structured Audit Logging That Doesn't Block the Critical Path

Audit logging has a performance trap. If your logger is synchronous and writes to a database, it adds 5-50ms to every request. That's unacceptable on the critical path. The right approach is to make logging asynchronous and use a fire-and-forget pattern — the request completes and the log entry is queued to be written separately.

We'll use asyncio.Queue as the buffer and a background task as the consumer. This decouples the request path from I/O latency completely:

# gateway/audit_logger.py
import asyncio
import json
import time
import logging
from dataclasses import dataclass, asdict
from typing import Optional
from gateway.models import GatewayRequest, GatewayResponse

logger = logging.getLogger(__name__)


@dataclass
class AuditRecord:
    request_id: str
    tenant_id: str
    user_id: str
    model: str
    provider: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cost_usd: float
    latency_ms: float
    cached: bool
    fallback_used: bool
    attempts: int
    success: bool
    error_message: Optional[str]
    timestamp: float
    metadata: dict


class AsyncAuditLogger:
    """
    Non-blocking audit logger. Writes to a queue and drains
    asynchronously via a background consumer task.
    """

    def __init__(
        self,
        sink,  # async callable that accepts list[AuditRecord]
        queue_size: int = 10_000,
        flush_interval_seconds: float = 5.0,
        batch_size: int = 100,
    ):
        self._sink = sink
        self._queue: asyncio.Queue[AuditRecord] = asyncio.Queue(maxsize=queue_size)
        self._flush_interval = flush_interval_seconds
        self._batch_size = batch_size
        self._consumer_task: Optional[asyncio.Task] = None
        self._dropped_count = 0

    async def start(self):
        """Start the background consumer. Call at gateway startup."""
        self._consumer_task = asyncio.create_task(self._consume_loop())

    async def stop(self):
        """Flush remaining records and stop. Call at gateway shutdown."""
        if self._consumer_task:
            self._consumer_task.cancel()
            try:
                await self._consumer_task
            except asyncio.CancelledError:
                pass
        # Drain remaining items
        remaining = []
        while not self._queue.empty():
            try:
                remaining.append(self._queue.get_nowait())
            except asyncio.QueueEmpty:
                break
        if remaining:
            await self._sink(remaining)

    def log_request(
        self,
        request: GatewayRequest,
        response: Optional[GatewayResponse] = None,
        error: Optional[Exception] = None,
    ) -> None:
        """
        Non-blocking. Puts record in queue or drops if full.
        Never raises an exception — logging must never break requests.
        """
        try:
            record = AuditRecord(
                request_id=request.request_id,
                tenant_id=request.tenant_id,
                user_id=request.user_id,
                model=request.model,
                provider=response.provider.value if response else "unknown",
                prompt_tokens=response.usage.prompt_tokens if response else 0,
                completion_tokens=response.usage.completion_tokens if response else 0,
                total_tokens=response.usage.total_tokens if response else 0,
                cost_usd=response.cost_usd if response else 0.0,
                latency_ms=response.latency_ms if response else 0.0,
                cached=response.cached if response else False,
                fallback_used=response.fallback_used if response else False,
                attempts=response.attempts if response else 0,
                success=error is None,
                error_message=str(error) if error else None,
                timestamp=time.time(),
                metadata=request.metadata,
            )
            self._queue.put_nowait(record)
        except asyncio.QueueFull:
            self._dropped_count += 1
            if self._dropped_count % 100 == 1:
                logger.error(
                    f"Audit log queue full! Dropped {self._dropped_count} records. "
                    "Increase queue_size or check sink performance."
                )
        except Exception:
            logger.exception("Unexpected error building audit record (swallowed)")

    async def _consume_loop(self):
        while True:
            batch = []
            deadline = time.monotonic() + self._flush_interval

            while time.monotonic() < deadline and len(batch) < self._batch_size:
                try:
                    remaining_time = deadline - time.monotonic()
                    if remaining_time <= 0:
                        break
                    record = await asyncio.wait_for(
                        self._queue.get(), timeout=remaining_time
                    )
                    batch.append(record)
                except asyncio.TimeoutError:
                    break

            if batch:
                try:
                    await self._sink(batch)
                except Exception:
                    logger.exception(f"Audit sink failed for {len(batch)} records")


# Example sink implementations

async def stdout_sink(records: list[AuditRecord]):
    """Development sink — just prints JSON."""
    for r in records:
        print(json.dumps(asdict(r)))


async def postgres_sink(pool, records: list[AuditRecord]):
    """Production sink — bulk insert into PostgreSQL."""
    if not records:
        return
    values = [
        (
            r.request_id, r.tenant_id, r.user_id, r.model, r.provider,
            r.prompt_tokens, r.completion_tokens, r.total_tokens,
            r.cost_usd, r.latency_ms, r.cached, r.fallback_used,
            r.attempts, r.success, r.error_message,
            r.timestamp, json.dumps(r.metadata)
        )
        for r in records
    ]
    async with pool.acquire() as conn:
        await conn.executemany(
            """
            INSERT INTO llm_audit_log (
                request_id, tenant_id, user_id, model, provider,
                prompt_tokens, completion_tokens, total_tokens,
                cost_usd, latency_ms, cached, fallback_used,
                attempts, success, error_message, timestamp, metadata
            ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,
                      to_timestamp($16),$17::jsonb)
            """,
            values,
        )

---

Assembling the Gateway Core

Now we wire everything together into a single LLMGateway class that represents the complete middleware stack:

# gateway/gateway.py
import asyncio
from gateway.models import GatewayRequest, GatewayResponse
from gateway.auth import GatewayAuthMiddleware, AuthError
from gateway.rate_limiter import RedisTokenBucketLimiter, RateLimitExceeded
from gateway.router import GatewayRouter
from gateway.audit_logger import AsyncAuditLogger
from gateway.circuit_breaker import CircuitBreakerOpen
import logging

logger = logging.getLogger(__name__)


class LLMGateway:
    def __init__(
        self,
        auth: GatewayAuthMiddleware,
        rate_limiter: RedisTokenBucketLimiter,
        router: GatewayRouter,
        audit_logger: AsyncAuditLogger,
    ):
        self.auth = auth
        self.rate_limiter = rate_limiter
        self.router = router
        self.audit_logger = audit_logger

    async def complete(
        self,
        request: GatewayRequest,
        auth_token: str,
    ) -> GatewayResponse:
        """
        Main entry point. Runs the full middleware stack:
        authenticate -> authorize -> rate limit -> route -> log.
        """
        response = None
        error = None

        try:
            # 1. Authenticate and attach identity
            identity = self.auth.authenticate(auth_token)
            request.tenant_id = identity.tenant_id
            request.user_id = identity.user_id

            # 2. Authorize model access
            self.auth.authorize_model(identity, request.model)

            # 3. Determine primary provider for rate limit check
            providers = self.router._get_provider_order(request)
            primary_provider = providers[0] if providers else None

            if primary_provider:
                await self.rate_limiter.check_request(
                    request, identity.rate_limit_tier, primary_provider
                )

            # 4. Route and execute
            response = await self.router.route(request)

            # 5. Record token usage for budget tracking (non-blocking)
            asyncio.create_task(
                self.rate_limiter.record_token_usage(
                    identity.tenant_id,
                    response.usage.total_tokens,
                    identity.rate_limit_tier,
                )
            )

            return response

        except (AuthError, RateLimitExceeded, CircuitBreakerOpen) as e:
            error = e
            raise

        except Exception as e:
            error = e
            logger.exception(f"Gateway error for request {request.request_id}")
            raise

        finally:
            # Logging always happens, success or failure
            self.audit_logger.log_request(request, response, error)

Notice the finally block — the audit log entry is written regardless of whether the request succeeded or failed. This is critical for compliance and debugging. A failed request that leaves no trace is a nightmare during an incident.

---

Exposing the Gateway as an HTTP Service

The gateway core is pure Python. To expose it as a service, we wrap it in a FastAPI application:

# gateway/api.py
from fastapi import FastAPI, HTTPException, Header, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from typing import Optional
from gateway.models import GatewayRequest, Message, MessageRole
from gateway.auth import AuthError
from gateway.rate_limiter import RateLimitExceeded
from gateway.circuit_breaker import CircuitBreakerOpen


app = FastAPI(title="LLM Gateway")


class ChatMessage(BaseModel):
    role: str
    content: str


class CompletionRequest(BaseModel):
    model: str
    messages: list[ChatMessage]
    max_tokens: int = 1024
    temperature: float = 0.7
    metadata: dict = {}


@app.post("/v1/chat/completions")
async def chat_completions(
    body: CompletionRequest,
    authorization: str = Header(...),
    gateway = None,  # injected via app.state
):
    # Strip "Bearer " prefix
    token = authorization.removeprefix("Bearer ").strip()

    request = GatewayRequest(
        messages=[
            Message(role=MessageRole(m.role), content=m.content)
            for m in body.messages
        ],
        model=body.model,
        max_tokens=body.max_tokens,
        temperature=body.temperature,
        metadata=body.metadata,
    )

    try:
        gw = app.state.gateway
        response = await gw.complete(request, token)
    except AuthError as e:
        raise HTTPException(status_code=401, detail=str(e))
    except RateLimitExceeded as e:
        return JSONResponse(
            status_code=429,
            content={"error": str(e), "retry_after": e.retry_after_seconds},
            headers={"Retry-After": str(int(e.retry_after_seconds))},
        )
    except CircuitBreakerOpen as e:
        raise HTTPException(status_code=503, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail="Gateway error")

    return {
        "id": response.request_id,
        "model": response.model,
        "provider": response.provider.value,
        "choices": [{"message": {"role": "assistant", "content": response.content}}],
        "usage": {
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
        },
        "gateway_metadata": {
            "latency_ms": response.latency_ms,
            "cost_usd": response.cost_usd,
            "fallback_used": response.fallback_used,
            "attempts": response.attempts,
        },
    }

---

Hands-On Exercise

Build on the gateway you've just implemented to complete this three-part exercise. Each part increases in complexity.

Part 1: Add a caching layer. Implement a SemanticCache class that, before routing a request, checks Redis for a response to the same (tenant, model, messages hash) combination. If found, return the cached response with cached=True and don't hit any provider. Use SHA-256 of the serialized messages as the cache key, with a configurable TTL (start with 3600 seconds). Make sure cached responses still go through the audit logger so you track cache hit rates.

Part 2: Implement cost-aware routing. Instead of a static routing table, implement a CostAwareRouter that, when multiple providers can handle a model, routes to the cheapest one first — unless the cheapest one's circuit is open. Add a cheapest_first flag to GatewayRequest that overrides the default routing table ordering. Write a unit test that verifies a request for a "smart" model routes to Anthropic when OpenAI costs more for that request size.

Part 3: Build an admin dashboard query. Write a SQL query against the llm_audit_log table that produces a daily cost report by tenant: total spend, total tokens, cache hit rate, fallback rate, and p95 latency. Then write a second query that identifies the top 10 most expensive individual requests in the past 7 days, including their metadata fields so you can identify which features generated them.

---

Common Mistakes & Troubleshooting

Mistake 1: Storing provider API keys in the canonical request This happens when developers pass the API key in the request payload to make testing easier, then forget to remove it. The consequence is that every audit log record contains a live API key. The fix is to never accept credentials from callers. Period. The credential store is the only place keys should live.

Mistake 2: Rate limiting only at the request level, not token level A request containing a 100,000-token prompt is not the same as a request with 100 tokens. If your rate limiter only counts requests, a single request can consume your entire daily budget. Always implement both request-per-minute and token-per-day limits.

Mistake 3: Circuit breaker that trips on rate limit errors We covered this in the routing section, but it deserves emphasis. If your circuit breaker opens on 429 responses, you'll start routing all traffic away from a perfectly healthy provider just because you exceeded your quota. Classify errors correctly: 429 = rate limited (skip this provider for now, try again), 5xx = potentially broken (record as circuit breaker failure).

Mistake 4: Synchronous audit logging Calling await db.insert(record) inside the request handler seems natural but adds 5-50ms to every request. At 100 RPS, this means your logging is consuming 500ms-5000ms of server time per second. Use the async queue pattern shown above, or use a dedicated log aggregation service (Datadog, Splunk, CloudWatch) and ship logs asynchronously.

Mistake 5: Not handling half-open state safely in concurrent scenarios In our circuit breaker implementation, the half-open state lets one test request through. But in a concurrent system, if 100 requests arrive simultaneously when the circuit transitions to half-open, all 100 might try to be that "test request." The correct fix is to use a lock (which we included via _half_open_lock) to serialize access during the half-open state.

Troubleshooting: Rate limit keys growing unboundedly If you don't set expiry on your Redis keys, the keyspace grows forever. Our Lua script includes EXPIRE 3600 calls, but you should also set a Redis maxmemory policy and monitor key count. redis-cli --scan --pattern "rl:*" | wc -l is a useful diagnostic command.

Troubleshooting: Audit log queue dropping records If you see the "Audit log queue full" warning, there are two possible causes: your sink is too slow (the consumer can't keep up), or your queue is sized too small for traffic spikes. Profile the sink first. If it's a database insert, try bulk inserting 500 records at once instead of 100. If the sink is already fast, increase queue_size. A queue size of 50,000 records at 1KB per record is only 50MB of memory.

---

Summary & Next Steps

You've built a production-grade LLM gateway from first principles. The architecture we implemented enforces a clean separation of concerns: adapters handle provider translation, the auth layer handles identity, the rate limiter enforces policies, the circuit breaker handles provider degradation, and the audit logger captures the full request lifecycle — all wired together by the gateway core.

The key design insights to carry forward:

• The canonical request/response model is the system's center of gravity. Protect its integrity.
• Every error type has a different remediation: auth errors fail fast, rate limits trigger 429s with retry headers, provider errors trigger fallback routing.
• The critical path (authenticate → rate limit → route → return) should contain zero synchronous I/O beyond the actual LLM call.
• Circuit breakers and rate limiters solve different problems and must not be conflated.

Where to go next:

• Streaming support: Modify the adapters to handle text/event-stream responses and propagate them through the gateway. This changes the response model significantly — you'll need to think about when to write audit log records when you don't have final token counts until the stream ends.
• Prompt injection detection: Add a middleware layer before routing that runs lightweight classifiers to detect prompt injection attempts. Projects like PromptGuard provide model-based detection you can integrate.
• Observability integration: Wire the gateway to emit OpenTelemetry traces, with spans for each middleware layer and the upstream provider call. This gives you distributed tracing across your entire system.
• Semantic caching with embeddings: Implement the caching exercise using vector similarity instead of exact hash matching. A request for "summarize this contract" and "give me a summary of this contract" should hit the same cache entry if the document content is identical.
• Multi-region deployment: When you run gateway instances across regions, your Redis state (rate limits, circuit breaker state) needs to be consistent. Explore Redis Cluster or a managed solution like Upstash for globally distributed rate limiting.