Your agent calls one model, and one bad night on that provider — a rash of 500s, a latency spike, a silent price change — takes your product down with it. This router is the fix in ~60 lines of Python: it tries your primary model, falls through an ordered list of backups on any error or timeout, refuses any call that would blow a per-request or daily budget, and quietly sends a slice of traffic to a cheap challenger while logging cost-per-task so you can decide with numbers instead of vibes.

It's a follow-up to the Terra vs Muse Spark vs Grok routing decision: that piece told you which cheap model to try; this one is the code that lets you try it in front of prod without holding your breath. The trick that makes it short is that they all speak the same wire format.

Step 1: One client shape for every provider#

Grok, Muse Spark, Terra, DeepSeek and most others expose an OpenAI-compatible /chat/completions endpoint. That means one client library — the standard openai Python package — talks to all of them, and the only thing that changes per provider is base_url and the API key. So a "route" is just a small record: a model id, an endpoint, which env var holds the key, and the two prices that will bill you.

from dataclasses import dataclass

@dataclass
class Route:
    name: str          # the provider's model id, e.g. "grok-4.5"
    base_url: str      # OpenAI-compatible endpoint
    api_key_env: str   # env var holding this provider's key
    in_price: float    # $ per 1M input tokens
    out_price: float   # $ per 1M output tokens

# All four speak the OpenAI Chat Completions format, so only base_url + key differ.
ROUTES = {
    "muse-spark": Route("muse-spark", "https://api.meta.ai/v1",    "META_API_KEY",     1.25,  4.25),
    "grok-4.5":   Route("grok-4.5",   "https://api.x.ai/v1",       "XAI_API_KEY",      2.00,  6.00),
    "terra":      Route("terra",      "https://api.openai.com/v1", "OPENAI_API_KEY",   2.50, 15.00),
    "deepseek":   Route("deepseek",   "https://api.deepseek.com",  "DEEPSEEK_API_KEY", 0.28,  0.42),
}

Prices are illustrative — put your real numbers in. The point is that adding a fifth provider is one line, not a new SDK.

Step 2: Get the cost math right#

This is the one line that has to be correct. Providers price per million tokens, and input and output are priced differently (output is usually 3–6x input), so the cost of a call is:

def cost_of(route: Route, in_tok: int, out_tok: int) -> float:
    # Load-bearing line. Prices are per-1M-tokens; input and output differ.
    return in_tok / 1e6 * route.in_price + out_tok / 1e6 * route.out_price

The subtlety that trips people up: you bill on the actual usage the API returns, not on an estimate. The response carries a usage block with prompt_tokens and completion_tokens (chat.completions reference). We only estimate tokens for the pre-flight budget check, where we don't have the real count yet.

Step 3: A hard cost cap#

Two budgets: a per-request ceiling (no single call should cost more than X) and a daily ceiling for the whole process. The per-request check runs before the call using a rough token estimate; the daily counter is updated after each call with the real cost.

import os, time, json, threading
from openai import OpenAI, APIError

PER_REQUEST_CAP = 0.05     # $ — refuse a single call projected above this
DAILY_CAP       = 25.00    # $ — hard stop for this process for the day

class BudgetError(Exception):
    """Raised when a call would breach a budget. NOT a provider failure."""

_spent = 0.0
_lock = threading.Lock()

def _add_spend(usd: float) -> float:
    global _spent
    with _lock:              # calls may run concurrently; keep the total honest
        _spent += usd
        return _spent

def _client(route: Route, timeout: float) -> OpenAI:
    # Per-client timeout + retries are the documented knobs on openai.OpenAI.
    return OpenAI(base_url=route.base_url, api_key=os.environ[route.api_key_env],
                  timeout=timeout, max_retries=2)

def _call(route: Route, messages, max_tokens: int, timeout: float):
    # Pre-flight cap. We don't know prompt tokens yet, so estimate ~4 chars/token
    # and assume the worst case: a full max_tokens reply.
    est_in = sum(len(m["content"]) for m in messages) // 4
    projected = cost_of(route, est_in, max_tokens)
    if projected > PER_REQUEST_CAP:
        raise BudgetError(f"{route.name}: projected ${projected:.4f} > ${PER_REQUEST_CAP}")
    if _spent >= DAILY_CAP:
        raise BudgetError(f"daily cap ${DAILY_CAP} reached (spent ${_spent:.2f})")

    t0 = time.perf_counter()
    resp = _client(route, timeout).chat.completions.create(
        model=route.name, messages=messages, max_tokens=max_tokens,
    )
    latency = time.perf_counter() - t0

    u = resp.usage                                          # ACTUAL tokens
    cost = cost_of(route, u.prompt_tokens, u.completion_tokens)
    total = _add_spend(cost)
    print(json.dumps({                                      # the row you group by later
        "model": route.name, "in": u.prompt_tokens, "out": u.completion_tokens,
        "cost_usd": round(cost, 6), "latency_s": round(latency, 3),
        "day_total_usd": round(total, 4),
    }))
    return resp.choices[0].message.content, cost

A budget breach raises BudgetError, which is deliberately not a provider failure — Step 4 treats the two differently. We put the cheapest model first in the fallback order so "refuse" only ever triggers when even the cheap option is too expensive; escalating a too-costly call to a pricier model would be exactly the wrong move.

Step 4: Fallbacks + a canary, in one function#

Now the router itself. On each request it decides whether this is a canary (a small % go to the challenger) or a normal request (primary), then appends the ordered fallback chain. It walks that list: a provider error (timeout, 5xx, rate limit — all subclasses of openai.APIError) means try the next route; a budget error means refuse and stop, because falling through would only spend more.

import random

PRIMARY    = "muse-spark"            # the cheap challenger you're promoting to primary
FALLBACKS  = ["grok-4.5", "terra"]   # ordered safety net, proven models
CHALLENGER = "deepseek"              # canary target
CANARY_PCT = 0.05                    # 5% of traffic

def route(messages, max_tokens: int = 512, timeout: float = 20.0) -> str:
    # Canary picks the head of the chain; the proven fallbacks always follow,
    # so a broken challenger can never take a request down — it just falls back.
    head = CHALLENGER if random.random() < CANARY_PCT else PRIMARY
    order = [head] + [m for m in FALLBACKS if m != head]

    last_err = None
    for name in order:
        try:
            text, _ = _call(ROUTES[name], messages, max_tokens, timeout)
            return text
        except BudgetError:
            raise                            # refuse — escalating would cost MORE
        except APIError as e:                # timeout / 5xx / rate limit
            last_err = e
            print(json.dumps({"fallthrough": name, "error": str(e)}))  # never silent
            continue
    raise RuntimeError(f"all routes exhausted: {last_err}")

if __name__ == "__main__":
    print(route([{"role": "user", "content": "Summarize the CAP theorem in one sentence."}]))

That's the whole thing: routes table, correct cost math, a two-tier budget, ordered fallbacks, and a logging canary — around 60 lines of logic. Every call emits one JSON row; pipe those to your logs and GROUP BY model to get real cost-per-task per model, which is the only honest way to compare cheap models on your own workload.

The TypeScript equivalent (same idea, openai npm)#

import OpenAI from "openai";

const cost = (r: Route, inTok: number, outTok: number) =>
  (inTok / 1e6) * r.inPrice + (outTok / 1e6) * r.outPrice;

async function callRoute(r: Route, messages: any[], maxTokens = 512) {
  const client = new OpenAI({ baseURL: r.baseUrl, apiKey: process.env[r.apiKeyEnv]!,
                              timeout: 20_000, maxRetries: 2 });
  const t0 = Date.now();
  const resp = await client.chat.completions.create({ model: r.name, messages, max_tokens: maxTokens });
  const u = resp.usage!;
  const c = cost(r, u.prompt_tokens, u.completion_tokens);
  console.log(JSON.stringify({ model: r.name, cost_usd: c, latency_ms: Date.now() - t0 }));
  return resp.choices[0].message.content ?? "";
}

Wrap callRoute in the same try/catch-and-fall-through loop and you have the JS version.

When to reach for a gateway instead#

Be honest about the ceiling here. This router is deliberately minimal, and two limits show up fast. First, the daily cap lives in one process's memory — run replicas and it under-counts by the replica count; the fix is a shared counter like Redis INCRBYFLOAT keyed to the day, or a gateway that governs spend centrally. Second, once you have many keys, many models, cooldowns, exponential backoff and per-tenant budgets, you're rebuilding a product that already exists.

That product is LiteLLM or OpenRouter. LiteLLM's Router gives you model-to-model fallbacks, retries, cooldowns, load balancing across keys and per-key spend limits behind the same OpenAI-compatible surface; OpenRouter does automatic provider failover behind one endpoint and one bill. If you want to see how the managed options stack up, we covered LiteLLM vs Portkey vs TensorZero. Hand-roll this router to understand and own the behavior with a handful of routes; adopt the gateway when the operational surface outgrows a script.

Wire it into your agent#

Dropping this in front of an existing agent loop is a one-line swap. Wherever your code currently does:

resp = client.chat.completions.create(model="muse-spark", messages=messages)
text = resp.choices[0].message.content

replace it with:

text = route(messages, max_tokens=512)

Your agent keeps calling one function; behind it, the primary is tried first, a real outage falls through to a proven model, a runaway request is refused before it bills you, and 5% of traffic is quietly measuring whether the challenger is actually cheaper on your tasks. Watch the fallthrough log lines — a rising rate is an incident, not the router working as designed, because the whole risk of fallback is that it makes a broken model look fine. Once the canary's cost-per-task numbers come in lower and stay stable, promote the challenger to PRIMARY, demote yesterday's primary into FALLBACKS, and you've swapped models in prod without a single anxious deploy.