Why this matters

AI applications that call a single provider directly have three problems: vendor lock-in, no fallback when the provider has an outage, and no mechanism to route cheap requests to cheap models. A model gateway solves all three. It gives your application one interface while routing calls intelligently across providers and models based on cost, latency, and capability requirements.

Core concepts

What a model gateway does

A model gateway is an abstraction layer between your application code and the raw provider SDKs. Your application calls gateway.complete(request). The gateway decides which provider and model to use, handles authentication, applies retry logic, and falls back to alternates when a provider fails.

Benefits:

• Portability — switch providers by changing gateway config, not application code
• Resilience — fallback chains survive provider outages automatically
• Cost control — route cheap requests (short, simple) to cheap models
• Observability — centralize logging, latency tracking, and cost accounting

Routing strategies

Cost-based routing — for each request, check whether a cheaper model can satisfy the capability requirements. Route to the cheapest that qualifies.

Latency-based routing — track rolling p95 latency per model. Route to the fastest that meets quality requirements.

Capability-based routing — some models support tools, vision, or long context. Route to the cheapest model that supports the required capabilities.

Model gateway implementation

from dataclasses import dataclass, field
from typing import Any
import time
import anthropic
import openai


@dataclass
class ModelConfig:
    provider: str          # "anthropic" | "openai"
    model_id: str
    cost_per_1k_tokens: float
    supports_tools: bool = False
    supports_vision: bool = False
    max_context_tokens: int = 8192


@dataclass
class GatewayRequest:
    messages: list[dict]
    system: str = ""
    max_tokens: int = 1024
    require_tools: bool = False
    require_vision: bool = False
    min_context_tokens: int = 0
    strategy: str = "cost"   # "cost" | "latency" | "capability"


@dataclass
class GatewayResponse:
    content: str
    model_used: str
    provider: str
    latency_ms: float
    attempt: int


class ModelGateway:
    def __init__(self, models: list[ModelConfig]):
        self.models = models
        self._latency_history: dict[str, list[float]] = {m.model_id: [] for m in models}
        self._anthropic = anthropic.Anthropic()
        self._openai = openai.OpenAI()

    def _eligible_models(self, req: GatewayRequest) -> list[ModelConfig]:
        return [
            m for m in self.models
            if (not req.require_tools or m.supports_tools)
            and (not req.require_vision or m.supports_vision)
            and m.max_context_tokens >= req.min_context_tokens
        ]

    def _rank_models(self, eligible: list[ModelConfig], strategy: str) -> list[ModelConfig]:
        if strategy == "cost":
            return sorted(eligible, key=lambda m: m.cost_per_1k_tokens)
        elif strategy == "latency":
            def avg_latency(m):
                hist = self._latency_history[m.model_id]
                return sum(hist[-10:]) / len(hist[-10:]) if hist else float("inf")
            return sorted(eligible, key=avg_latency)
        return eligible

    def _call_provider(self, model: ModelConfig, req: GatewayRequest) -> str:
        if model.provider == "anthropic":
            resp = self._anthropic.messages.create(
                model=model.model_id,
                max_tokens=req.max_tokens,
                system=req.system,
                messages=req.messages,
            )
            return resp.content[0].text
        elif model.provider == "openai":
            messages = ([{"role": "system", "content": req.system}] if req.system else []) + req.messages
            resp = self._openai.chat.completions.create(
                model=model.model_id,
                max_tokens=req.max_tokens,
                messages=messages,
            )
            return resp.choices[0].message.content
        raise ValueError(f"Unknown provider: {model.provider}")

    def complete(self, req: GatewayRequest) -> GatewayResponse:
        eligible = self._eligible_models(req)
        if not eligible:
            raise RuntimeError("No models satisfy the request requirements")
        ranked = self._rank_models(eligible, req.strategy)

        for attempt, model in enumerate(ranked, start=1):
            start = time.perf_counter()
            try:
                content = self._call_provider(model, req)
                latency_ms = (time.perf_counter() - start) * 1000
                self._latency_history[model.model_id].append(latency_ms)
                return GatewayResponse(content, model.model_id, model.provider, latency_ms, attempt)
            except Exception as exc:
                latency_ms = (time.perf_counter() - start) * 1000
                print(f"[gateway] {model.model_id} failed (attempt {attempt}): {exc}")
                continue

        raise RuntimeError(f"All {len(ranked)} models failed for request")

Using the gateway

gateway = ModelGateway([
    ModelConfig("anthropic", "claude-3-5-haiku-20241022", cost_per_1k_tokens=0.001, supports_tools=True),
    ModelConfig("anthropic", "claude-opus-4-5", cost_per_1k_tokens=0.015, supports_tools=True),
    ModelConfig("openai", "gpt-4o-mini", cost_per_1k_tokens=0.00015, supports_vision=True),
])

# Simple request — routes to cheapest eligible model
resp = gateway.complete(GatewayRequest(
    messages=[{"role": "user", "content": "Summarize this in one sentence: ..."}],
    strategy="cost",
))
print(f"Used {resp.model_used} in {resp.latency_ms:.0f}ms (attempt {resp.attempt})")

# Tool-requiring request — haiku or opus only, never gpt-4o-mini
resp = gateway.complete(GatewayRequest(
    messages=[{"role": "user", "content": "Search for recent papers on RAG."}],
    require_tools=True,
    strategy="cost",
))

Common mistakes

No fallback chain — calling one provider with no fallback means an outage takes your feature down. Always configure at least two providers.

Tracking latency but not weighting recency — the last 10 calls matter more than calls from 6 hours ago. Use a rolling window or exponential moving average, not a full history mean.

Routing all requests to the cheapest model — some tasks genuinely require a stronger model. Capability-based routing prevents the gateway from sending complex reasoning tasks to models that will produce low-quality output.

Missing observability — the gateway is the right place to log model_used, latency, token counts, and cost per request. Without this data you cannot tune routing rules or catch cost anomalies.

Try it yourself

Build the ModelGateway class above with at least two real or mocked providers. Write a test that: (1) configures three model configs with different costs, (2) sends a request with strategy="cost" and verifies the cheapest eligible model was used, (3) makes that model raise an exception and verifies the gateway falls back to the second cheapest, (4) sends a request with require_tools=True and verifies only tool-capable models are in the candidate set. Log which model was used and the attempt number for each call.