Step 1: Pass-through: one door, and the model behind it
Northwind's three apps call the model with one shared provider key, from three codebases, with three different ideas about retries. The bill doubled last month and nobody can say which app did it. The fix is one door: a gateway every app calls instead of the provider, so that everything you add in the next seven steps (keys, a log, prices, limits, a cache, retries, routing) happens once, in one place, for all of them.
Open gateway.py. Gateway.handle() is already the full pipeline: it
authenticates, routes, checks the idempotency store and the cache, applies
limits, calls the upstream with retries and a fallback, prices the answer,
logs a record and responds. Every stage delegates to a small method, and
most of those methods are stubs that later steps fill in. Right now the
gateway cannot even reach the model, because call_once() is the stub for
this step.
upstream.py gives you two upstreams with the same interface. Upstream
talks to the lab's model proxy over HTTP. FakeUpstream answers instantly
with deterministic text and token counts, can be scripted to fail, and
moves a fake clock, so the failure, limit and cache steps are exact and
free. server.py puts Gateway.handle() behind a plain HTTP server, and
client.py is the app: it sends a chat completion and prints the status,
the gateway's response headers and the answer.
do this
1. Implement Gateway.call_once(body, model). Copy the request body,
replace its model with the model the router resolved (clients send
aliases such as fast and smart; the upstream needs the real id) and
return self.upstream.chat(payload). Errors propagate: the retry step will
catch them.
2. Run. The first command starts the gateway against the live model
and sends one question through it. The second starts it with --fake and
sends the same question to the fake upstream. Both print HTTP 200, an
x-request-id, an x-model-served header, and an answer. Notice the
served name: the provider reports meta-llama/llama-3.1-8b-instruct for a
request that asked for meta/llama-3.1-8b-instruct. Step 3 prices on the
served name for exactly that reason.
3. Read handle() once, top to bottom. You will not change it. Each
later step fills in one of the methods it calls, and the order of those
calls is the design: the cheapest checks first, the upstream call last.
Why this matters in practice
Gateways are sold as products with long feature lists. Every one of those features is a method in this file. The pipeline order is the part vendors rarely explain: authentication before anything else, the cache before the rate limiter (a cached answer costs nothing, so it should not spend a slot), the limiter before the upstream call (the point is to stop spend before it happens), pricing and logging after, on every path, including the failures.
gateway.py, the file you edit225 lines
"""gateway.py: one door between Northwind's apps and the model.
python3 server.py --port 8088 # real upstream (the lab's model proxy)
python3 server.py --port 8088 --fake # deterministic fake upstream
python3 client.py --key nw-sk-internal-91c0 --model fast "What does a gateway do?"
Every request goes through Gateway.handle(): authenticate -> route -> idempotency ->
cache -> limits -> upstream (with retries and fallback) -> price -> log -> respond.
"""
import hashlib
import json
import random
import time
import uuid
from collections import deque
from dataclasses import dataclass, field
import yaml
from upstream import UpstreamError, UpstreamTimeout, estimate_tokens
MODELS = {"fast": "meta/llama-3.1-8b-instruct", "smart": "meta/llama-3.3-70b-instruct"}
FALLBACKS = {MODELS["fast"]: MODELS["smart"], MODELS["smart"]: MODELS["fast"]}
RETRYABLE = {408, 429, 500, 502, 503, 504}
ANONYMOUS = {"key": None, "user": None, "app": None, "tier": "open", "rpm": None,
"daily_budget_usd": None, "models": None, "active": True}
@dataclass
class Request:
method: str
path: str
headers: dict = field(default_factory=dict) # lower-case header names
body: dict = None
@dataclass
class Response:
status: int
body: dict
headers: dict = field(default_factory=dict)
def error(status, kind, message, headers=None):
return Response(status, {"error": {"type": kind, "message": message}}, dict(headers or {}))
def load_keys(path="keys.json"):
keys = json.load(open(path))
for k, rec in keys.items():
rec["key"] = k
return keys
def load_pricing(path="pricing.yaml"):
return yaml.safe_load(open(path))
class Gateway:
def __init__(self, upstream, keys=None, pricing=None, log_path="gateway.jsonl", cache_ttl=300,
max_attempts=3, base_delay=0.2, auto_threshold_tokens=800,
clock=time.time, sleep=time.sleep, rng=None):
self.upstream = upstream
self.keys = keys if keys is not None else load_keys()
self.pricing = pricing if pricing is not None else load_pricing()
self.log_path = log_path
self.cache_ttl = cache_ttl
self.max_attempts = max_attempts
self.base_delay = base_delay
self.auto_threshold_tokens = auto_threshold_tokens
self.clock, self.sleep, self.rng = clock, sleep, (rng or random.Random())
self.windows = {} # key -> deque of upstream-call timestamps (rate limit window)
self.spend = {} # key -> {"day": int, "usd": float}
self.cache = {} # cache key -> {"ts": float, "data": dict}
self.idempotent = {} # (key, idempotency-key) -> Response
# ------------------------------------------------------------------ entry point
def handle(self, req):
t0 = self.clock()
rec = {"request_id": uuid.uuid4().hex[:12], "ts": round(t0, 3), "method": req.method, "path": req.path,
"user": None, "app": None, "model": None, "served": None, "served_by": None, "status": None,
"latency_ms": None, "prompt_tokens": 0, "completion_tokens": 0, "cost_usd": 0.0,
"cache": None, "attempts": 0, "route": None, "error": None}
try:
resp = self._dispatch(req, rec)
except NotImplementedError as e:
resp = error(501, "not_implemented", str(e))
except Exception as e: # a gateway must answer even when it is broken
resp = error(500, "internal", f"{type(e).__name__}: {e}")
rec["status"] = resp.status
rec["latency_ms"] = round((self.clock() - t0) * 1000, 1)
if resp.status >= 400 and not rec["error"]:
rec["error"] = resp.body.get("error", {}).get("type", "error")
resp.headers.setdefault("x-request-id", rec["request_id"])
self.log(rec)
return resp
def _dispatch(self, req, rec):
if req.method == "GET" and req.path == "/health":
return Response(200, {"status": "ok"})
key = self.authenticate(req.headers)
if key is None:
return error(401, "auth", "missing, unknown or revoked API key; send Authorization: Bearer <key>")
rec["user"], rec["app"] = key["user"], key["app"]
if req.method == "GET" and req.path == "/v1/usage":
return Response(200, self.usage(key))
if req.method != "POST" or req.path != "/v1/chat/completions":
return error(404, "not_found", f"{req.method} {req.path} is not served by this gateway")
body = req.body
if not isinstance(body, dict) or not isinstance(body.get("messages"), list) or not body["messages"]:
return error(400, "invalid_request", "body must be a JSON object with a non-empty messages list")
if body.get("stream"):
return error(400, "invalid_request", "this gateway does not stream; remove stream from the request")
rec["model"] = body.get("model") or "auto"
route = self.route(body, key)
if isinstance(route, Response):
return route
rec["route"] = route
model = route["primary"]
idem = req.headers.get("idempotency-key")
if idem:
replay = self.idempotent.get((key["key"], idem))
if replay is not None:
rec["cache"], rec["served"] = "idempotent", replay.headers.get("x-model-served")
return Response(replay.status, replay.body, dict(replay.headers, **{"x-idempotent-replay": "true"}))
ck = self.cache_key(body, model)
bypass = "no-cache" in req.headers.get("cache-control", "")
hit = None if bypass else self.cache_get(ck, self.clock())
if hit is not None:
rec["cache"], rec["served"] = "hit", hit.get("model")
return Response(200, hit, {"x-cache": "HIT", "x-cost-usd": "0.00000000",
"x-model-served": hit.get("model", ""), "x-gateway-user": key["user"]})
rec["cache"] = "bypass" if bypass else "miss"
limited = self.check_limits(key, self.clock())
if limited is not None:
return limited
status, data, attempts = self.call_with_retries(body, model)
rec["attempts"], served_by = attempts, "primary"
pass # the fallback model arrives in Step 07
if status != 200:
rec["error"] = "upstream"
msg = data.get("error", {}).get("message") if isinstance(data, dict) else None
return error(502, "upstream", f"upstream returned {status}" + (f": {msg}" if msg else ""),
{"x-upstream-status": str(status), "x-upstream-attempts": str(rec["attempts"])})
usage = data.get("usage") or {}
served = data.get("model") or model
pt, ct = usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0)
cost = self.price(served, usage)
rec.update(served=served, served_by=served_by, prompt_tokens=pt, completion_tokens=ct, cost_usd=cost)
self.record_spend(key, cost, self.clock())
self.cache_put(ck, data, self.clock())
resp = Response(200, data, {"x-cache": "BYPASS" if bypass else "MISS", "x-model-served": served,
"x-served-by": served_by, "x-cost-usd": f"{cost:.8f}", "x-tokens": f"{pt}+{ct}",
"x-upstream-attempts": str(rec["attempts"]), "x-gateway-user": key["user"]})
if idem:
self.idempotent[(key["key"], idem)] = resp
return resp
# ------------------------------------------------------------------ Step 1: the call
def call_once(self, body, model):
"""One upstream call with `model` substituted for whatever alias the client sent.
Returns the upstream JSON; raises UpstreamError / UpstreamTimeout."""
# TODO (Step 01): payload = dict(body, model=model); return self.upstream.chat(payload)
raise NotImplementedError("Step 01: implement Gateway.call_once()")
# ------------------------------------------------------------------ Step 2: keys and the log
def authenticate(self, headers):
"""Return the key record for `Authorization: Bearer <key>`, or None."""
return dict(ANONYMOUS) # everyone is anonymous until Step 02
def log(self, rec):
"""Append one JSON line per request to self.log_path."""
pass # the request log arrives in Step 02
# ------------------------------------------------------------------ Step 3: price and account
def price(self, model, usage):
"""USD for one call: tokens / 1e6 * the per-million rate for the SERVED model name."""
return 0.0 # pricing arrives in Step 03
def record_spend(self, key, cost, now):
"""Add `cost` to the key's spend for today (UTC day = int(now // 86400))."""
pass # spend accounting arrives in Step 03
def usage(self, key):
"""What GET /v1/usage returns for the calling key."""
return {"user": key["user"], "app": key["app"], "note": "usage accounting arrives in Step 03"}
# ------------------------------------------------------------------ Step 4: limits and budgets
def check_limits(self, key, now):
"""None if the call may proceed, else a 429 Response. Counts only calls that reach the upstream."""
return None # limits arrive in Step 04
# ------------------------------------------------------------------ Step 5: exact-match cache
def cache_key(self, body, model):
"""sha256 of the canonical JSON of everything that changes the answer."""
return None # caching arrives in Step 05
def cache_get(self, ck, now):
return None # caching arrives in Step 05
def cache_put(self, ck, data, now):
pass # caching arrives in Step 05
# ------------------------------------------------------------------ Step 6: retries
def call_with_retries(self, body, model):
"""(status, data, attempts). Retries RETRYABLE statuses and timeouts with exponential
backoff and jitter, honours retry-after, never retries other 4xx."""
try:
return 200, self.call_once(body, model), 1
except UpstreamTimeout as e:
return 504, {"error": {"type": "upstream_timeout", "message": str(e)}}, 1
except UpstreamError as e:
return e.status, e.body, 1
# ------------------------------------------------------------------ Step 7: routing
def route(self, body, key):
"""{"requested", "primary", "fallback", "reason"} or a 403 Response."""
requested = body.get("model") or "auto"
return {"requested": requested, "primary": MODELS.get(requested, requested), "fallback": None, "reason": "passthrough"}provided for you:client.pykeys.jsonpricing.yamlserver.pyupstream.py
