An Anti-Corruption Layer for Claude API Models — Keeping Generation Changes Out of Your Business Logic

// model-registry.ts — the one and only place that confines "external circumstances"
export type ModelRole = "fast" | "balanced" | "deep";
 
interface ModelBinding {
  primary: string;        // the physical model ID normally used
  fallbacks: string[];    // the order to descend through on failure or retirement
  inputPer1M: number;     // USD per 1M input tokens
  outputPer1M: number;    // USD per 1M output tokens
  maxOutputTokens: number;
}
 
// rewriting only this completes a model-generation change across every app
const REGISTRY: Record<ModelRole, ModelBinding> = {
  fast: {
    primary: "claude-haiku-4-5-20251001",
    fallbacks: ["claude-3-5-haiku-latest"],
    inputPer1M: 1.0,
    outputPer1M: 5.0,
    maxOutputTokens: 4096,
  },
  balanced: {
    primary: "claude-sonnet-4-6",
    fallbacks: ["claude-haiku-4-5-20251001"],
    inputPer1M: 3.0,
    outputPer1M: 15.0,
    maxOutputTokens: 8192,
  },
  deep: {
    primary: "claude-opus-4-6",
    fallbacks: ["claude-sonnet-4-6"],
    inputPer1M: 15.0,
    outputPer1M: 75.0,
    maxOutputTokens: 16384,
  },
};
 
export function resolve(role: ModelRole): ModelBinding {
  const binding = REGISTRY[role];
  if (!binding) throw new Error(`unknown role: ${role}`);
  return binding;
}
 
export function chain(role: ModelRole): string[] {
  const b = resolve(role);
  return [b.primary, ...b.fallbacks];
}

// model-gateway.ts — business code needs to know only this ask()
import Anthropic from "@anthropic-ai/sdk";
import { resolve, chain, type ModelRole } from "./model-registry";
 
const client = new Anthropic();
 
interface AskResult {
  text: string;
  modelUsed: string;
  usdCost: number;
  degraded: boolean;   // did we drop below primary?
}
 
export async function ask(
  role: ModelRole,
  prompt: string,
  opts: { maxTokens?: number } = {},
): Promise<AskResult> {
  const binding = resolve(role);
  const candidates = chain(role);
  let lastErr: unknown;
 
  for (let i = 0; i < candidates.length; i++) {
    const model = candidates[i];
    try {
      const res = await client.messages.create({
        model,
        max_tokens: opts.maxTokens ?? binding.maxOutputTokens,
        messages: [{ role: "user", content: prompt }],
      });
      const inTok = res.usage.input_tokens;
      const outTok = res.usage.output_tokens;
      // fallbacks may have different pricing, so recalculate on the model actually used
      const usdCost = priceFor(model, inTok, outTok);
      const text = res.content
        .filter((b) => b.type === "text")
        .map((b) => (b as { text: string }).text)
        .join("");
      return { text, modelUsed: model, usdCost, degraded: i > 0 };
    } catch (err) {
      lastErr = err;
      // treat retirement (404 model_not_found) and overload (529) identically as "descend"
      continue;
    }
  }
  throw new Error(`all candidates failed for role=${role}: ${String(lastErr)}`);
}
 
function priceFor(model: string, inTok: number, outTok: number): number {
  // a real-model price table not tied to roles (covers fallbacks too)
  const table: Record<string, [number, number]> = {
    "claude-haiku-4-5-20251001": [1.0, 5.0],
    "claude-3-5-haiku-latest": [0.8, 4.0],
    "claude-sonnet-4-6": [3.0, 15.0],
    "claude-opus-4-6": [15.0, 75.0],
  };
  const [pin, pout] = table[model] ?? [0, 0];
  return (inTok / 1e6) * pin + (outTok / 1e6) * pout;
}

// review-summary.ts — the caller knows only the role
import { ask } from "./model-gateway";
 
export async function summarizeReview(text: string) {
  const { text: summary, degraded } = await ask(
    "fast",
    `Summarize the following app review in one sentence:\n${text}`,
  );
  if (degraded) {
    // on degradation, just log; never stop the user experience
    console.warn("[review-summary] running on fallback model");
  }
  return summary;
}

# model_gateway.py — drawing the same boundary in Python as the TypeScript version
from dataclasses import dataclass
from anthropic import Anthropic
 
client = Anthropic()
 
@dataclass(frozen=True)
class Binding:
    primary: str
    fallbacks: tuple[str, ...]
    max_output: int
 
REGISTRY: dict[str, Binding] = {
    "fast":     Binding("claude-haiku-4-5-20251001", ("claude-3-5-haiku-latest",), 4096),
    "balanced": Binding("claude-sonnet-4-6", ("claude-haiku-4-5-20251001",), 8192),
    "deep":     Binding("claude-opus-4-6", ("claude-sonnet-4-6",), 16384),
}
 
def ask(role: str, prompt: str, max_tokens: int | None = None) -> dict:
    b = REGISTRY[role]
    candidates = (b.primary, *b.fallbacks)
    last_err: Exception | None = None
    for i, model in enumerate(candidates):
        try:
            res = client.messages.create(
                model=model,
                max_tokens=max_tokens or b.max_output,
                messages=[{"role": "user", "content": prompt}],
            )
            text = "".join(blk.text for blk in res.content if blk.type == "text")
            return {"text": text, "model_used": model, "degraded": i > 0}
        except Exception as err:  # treat retirement and overload identically as "next"
            last_err = err
            continue
    raise RuntimeError(f"all candidates failed for role={role}: {last_err}")

# test_model_contract.py — ping each role's primary daily to confirm it exists and responds
import pytest
from model_gateway import REGISTRY, client
 
@pytest.mark.parametrize("role", list(REGISTRY.keys()))
def test_primary_model_alive(role):
    model = REGISTRY[role].primary
    res = client.messages.create(
        model=model,
        max_tokens=8,
        messages=[{"role": "user", "content": "ping"}],
    )
    assert res.stop_reason in ("end_turn", "max_tokens")