Building a Provider-Agnostic LLM Client: One Interface for Claude, GPT, Gemini, and Mistral | AI Error Hub
AI Error Hub/Blog/Provider-Agnostic LLM Client
Tutorial · Code

Building a Provider-Agnostic LLM Client: One Interface for Claude, GPT, Gemini, and Mistral

Every team that runs more than one model provider ends up writing the same adapter layer, usually in a hurry after an outage. This post builds it on purpose: one Python interface over Claude, GPT, Gemini, and Mistral with normalized messages, tools, structured output, errors, retries, fallback, and cost tracking. It is a few hundred lines, uses no framework, and every piece can be tested without calling a real API.

ZK
Zain Khan
Editor, AI Error Hub · October 13, 2026 · 16 min read

Should you build this or buy it?

Gateways such as LiteLLM, OpenRouter, and Portkey already solve most of this, and for many teams they are the right answer. Our gateway comparison covers the tradeoffs in detail.

Build your own when you need exact control over which errors retry and which trigger failover, when requests cannot pass through a third party, or when you want a dependency surface you can read in one sitting. Buy when you need dozens of providers, hosted dashboards, per-key budgets, and you would rather not maintain adapters as four APIs change underneath you. Even if you buy, the design rules below are a good checklist for evaluating what you are buying.

Six design rules

  1. Normalize to the smallest useful core. Text, tool calls, JSON output, usage, and a stop reason. Nothing else belongs in the shared types.
  2. Errors are part of the interface. Callers catch your exception classes and never an SDK's.
  3. Store history in your own format. If conversations are saved in a provider's shape, you cannot fail over mid-conversation. If they are saved in yours, you can.
  4. Leak the other 20 percent on purpose. Caching, extended thinking, and grounding differ too much to unify. Give them a per-provider options escape hatch instead of pretending.
  5. Routing is data. Application code asks for "smart" or "cheap". Config decides which providers and models answer, and in what order.
  6. Every adapter passes the same contract tests. A new provider is done when the suite is green, not when it looks right.

What the adapters have to absorb

The four providers agree on the idea and disagree on nearly every detail. This table is the whole reason adapters exist:

ClaudeGPTGeminiMistral
System prompttop-level systemfirst message, role systemsystem_instruction in configfirst message, role system
Assistant role nameassistantassistantmodelassistant
Tool arguments arrive asparsed objectJSON stringparsed objectJSON string
Tool result matched bytool_use_idtool_call_idfunction nametool_call_id
Truncation signalmax_tokenslengthMAX_TOKENSlength
Output cap parametermax_tokens (required)max_completion_tokensmax_output_tokensmax_tokens
Overload status529503503503

Step 1: The normalized types

These dataclasses are the only types your application ever touches. Note that Message carries tool calls and tool results as structured fields, which is what keeps history portable.

from dataclasses import dataclass, field
from typing import Any, Literal, Optional

@dataclass
class ToolSpec:
    name: str
    description: str
    parameters: dict                 # plain JSON Schema

@dataclass
class ToolCall:
    id: str
    name: str
    arguments: dict

@dataclass
class Message:
    role: Literal["user", "assistant", "tool"]
    text: str = ""
    tool_calls: list[ToolCall] = field(default_factory=list)
    tool_call_id: Optional[str] = None       # set when role == "tool"

@dataclass
class Request:
    model: str                               # a logical alias: "smart", "cheap"
    messages: list[Message]
    system: str = ""
    tools: list[ToolSpec] = field(default_factory=list)
    max_tokens: int = 1024
    temperature: float = 0.0
    json_schema: Optional[dict] = None
    options: dict[str, dict] = field(default_factory=dict)   # per-provider escape hatch

@dataclass
class Usage:
    input_tokens: int = 0
    output_tokens: int = 0
    cached_tokens: int = 0

@dataclass
class Response:
    text: str
    tool_calls: list[ToolCall]
    stop: Literal["end", "length", "tool_use", "refusal", "other"]
    usage: Usage
    provider: str
    model: str
    data: Optional[dict] = None              # parsed JSON when json_schema was set
    attempts: int = 1
    fell_back_from: list[str] = field(default_factory=list)
    cost_usd: float = 0.0

Refusals are a stop reason, not an exception. A refused request returns HTTP 200 on most providers, and treating it as a normal outcome forces callers to handle it deliberately.

Step 2: Errors are part of the interface

Each error class carries two flags that drive all later behavior: whether retrying the same provider can help, and whether another provider might succeed. One function maps an HTTP status to the right class, so the adapters stay small.

class LLMError(Exception):
    retryable = False        # same provider, try again
    fallback = False         # a different provider might work

    def __init__(self, message, retry_after=None):
        super().__init__(message)
        self.retry_after = retry_after

class RateLimited(LLMError):
    retryable = True
    fallback = True

class Overloaded(LLMError):
    retryable = True
    fallback = True

class Transient(LLMError):          # timeouts, 5xx, connection resets
    retryable = True
    fallback = True

class AuthFailed(LLMError):         # alert loudly, but another provider may be fine
    fallback = True

class BadRequest(LLMError):         # the same request will fail everywhere
    pass

def from_status(status, message, retry_after=None, code=None):
    if status == 429 and code == "insufficient_quota":
        return AuthFailed(message)             # a billing problem: retrying cannot help
    if status == 429:
        return RateLimited(message, retry_after)
    if status in (503, 529):
        return Overloaded(message)
    if status in (401, 403):
        return AuthFailed(message)
    if status in (400, 404, 422):
        return BadRequest(message)
    if status >= 500:
        return Transient(message)
    return BadRequest(message)

def retry_after(response):
    try:
        return float(response.headers.get("retry-after"))
    except (TypeError, ValueError, AttributeError):
        return None

Two details are easy to miss. OpenAI returns 429 with the code insufficient_quota when you run out of credit, and no amount of backoff fixes that, so it is mapped to a non-retryable class. And Anthropic's overload status is 529, not 503, so a classifier that only knows 503 lets Claude capacity events slip through as hard failures. The retry theory behind these flags is in the Retry Logic Patterns guide.

If you want to route oversized prompts to a larger-context model, add a ContextTooLong class and detect it from the provider's error message text, which is the only signal some providers give.

Step 3: Schema helpers

Write each JSON Schema once, in plain draft-style JSON Schema, and let the adapters reshape it. The three dialects differ in small but fatal ways: OpenAI strict mode demands every key be listed as required and every object closed, Anthropic and Mistral want closed objects, and Gemini's schema subset has historically rejected additionalProperties.

import copy

def walk(schema, fn):
    schema = copy.deepcopy(schema)
    def go(node):
        if isinstance(node, dict):
            fn(node)
            for value in node.values():
                go(value)
        elif isinstance(node, list):
            for item in node:
                go(item)
    go(schema)
    return schema

def closed(schema):                 # Anthropic, Mistral: no extra keys
    def fn(node):
        if node.get("type") == "object":
            node["additionalProperties"] = False
    return walk(schema, fn)

def openai_strict(schema):          # OpenAI strict mode: all keys required, no extras
    def fn(node):
        if node.get("type") == "object":
            node["additionalProperties"] = False
            node["required"] = list(node.get("properties", {}).keys())
    return walk(schema, fn)

def gemini_safe(schema):            # drop keywords Gemini's subset may reject
    def fn(node):
        node.pop("additionalProperties", None)
        node.pop("$schema", None)
    return walk(schema, fn)

Because strict mode forces every key to be required, mark genuinely optional fields as nullable in your canonical schema, for example {"type": ["string", "null"]}. Otherwise the model is pushed to invent a value. Newer Gemini SDK versions can accept raw JSON Schema through a separate field, so check your version before assuming the stripping is still needed. Our Structured Output Reliability guide covers why schema-valid output can still be wrong.

Step 4: The adapter contract and the Claude adapter

An adapter is any object with a name and an async complete(model_id, req) that returns a Response or raises an LLMError. Here is Claude's. Retries are disabled in the SDK on purpose, so there is exactly one retry layer in the system.

import anthropic

STOP_MAP = {"end_turn": "end", "stop_sequence": "end", "max_tokens": "length",
            "tool_use": "tool_use", "refusal": "refusal"}

class AnthropicAdapter:
    name = "anthropic"

    def __init__(self, api_key=None):
        self.client = anthropic.AsyncAnthropic(api_key=api_key, max_retries=0)

    def _messages(self, msgs):
        out = []
        for m in msgs:
            if m.role == "tool":
                block = {"type": "tool_result", "tool_use_id": m.tool_call_id, "content": m.text}
                prev = out[-1] if out else None
                if (prev and prev["role"] == "user" and isinstance(prev["content"], list)
                        and prev["content"][-1]["type"] == "tool_result"):
                    prev["content"].append(block)     # parallel results share one user turn
                else:
                    out.append({"role": "user", "content": [block]})
            elif m.role == "assistant":
                blocks = [{"type": "text", "text": m.text}] if m.text else []
                for c in m.tool_calls:
                    blocks.append({"type": "tool_use", "id": c.id, "name": c.name, "input": c.arguments})
                out.append({"role": "assistant", "content": blocks})
            else:
                out.append({"role": "user", "content": m.text})
        return out

    async def complete(self, model_id, req):
        tools = [{"name": t.name, "description": t.description, "input_schema": t.parameters}
                 for t in req.tools]
        kwargs = dict(model=model_id, max_tokens=req.max_tokens,
                      temperature=min(req.temperature, 1.0),
                      messages=self._messages(req.messages))
        if req.system:
            kwargs["system"] = req.system
        if req.json_schema:                           # structured output = a forced tool call
            tools.append({"name": "emit_result",
                          "description": "Return the final structured result.",
                          "input_schema": closed(req.json_schema)})
            kwargs["tool_choice"] = {"type": "tool", "name": "emit_result"}
        if tools:
            kwargs["tools"] = tools
        kwargs.update(req.options.get(self.name, {}))
        try:
            r = await self.client.messages.create(**kwargs)
        except anthropic.APIStatusError as e:
            raise from_status(e.status_code, str(e), retry_after(e.response)) from e
        except (anthropic.APITimeoutError, anthropic.APIConnectionError) as e:
            raise Transient(str(e)) from e

        text = "".join(b.text for b in r.content if b.type == "text")
        calls = [ToolCall(b.id, b.name, b.input) for b in r.content if b.type == "tool_use"]
        stop = STOP_MAP.get(r.stop_reason, "other")
        data = None
        if req.json_schema:
            forced = [c for c in calls if c.name == "emit_result"]
            data = forced[0].arguments if forced else None
            calls = [c for c in calls if c.name != "emit_result"]
            if data is not None:
                stop = "end"
        usage = Usage(r.usage.input_tokens, r.usage.output_tokens,
                      getattr(r.usage, "cache_read_input_tokens", 0) or 0)
        return Response(text, calls, stop, usage, self.name, model_id, data)

Notice what the adapter does that callers never see: it merges parallel tool results into one user turn (Claude rejects them otherwise), clamps temperature to Claude's range, and turns a structured-output request into a forced tool call. Because the forced call means the model cannot also call your real tools, treat json_schema requests as extraction calls, not agent turns.

Step 5: OpenAI and Mistral share one adapter

Mistral's chat endpoint is OpenAI-shaped, close enough that the OpenAI SDK works against it for chat, tools, and JSON schema. So the Mistral adapter is a subclass that changes the base URL, the output-cap parameter, and the schema dialect. Use the official mistralai SDK instead if you need Mistral-only features, and verify a feature against current docs before relying on it.

import json, openai

FINISH_MAP = {"stop": "end", "length": "length", "tool_calls": "tool_use",
              "content_filter": "refusal"}
NO_TEMPERATURE = {"o5"}             # reasoning models that reject a temperature setting

def as_dict(value):
    return value if isinstance(value, dict) else json.loads(value or "{}")

class OpenAIAdapter:
    name = "openai"
    token_param = "max_completion_tokens"
    schema_fn = staticmethod(openai_strict)

    def __init__(self, api_key=None, base_url=None):
        self.client = openai.AsyncOpenAI(api_key=api_key, base_url=base_url, max_retries=0)

    def _messages(self, req):
        out = [{"role": "system", "content": req.system}] if req.system else []
        for m in req.messages:
            if m.role == "tool":
                out.append({"role": "tool", "tool_call_id": m.tool_call_id, "content": m.text})
            elif m.role == "assistant":
                msg = {"role": "assistant", "content": m.text or None}
                if m.tool_calls:
                    msg["tool_calls"] = [
                        {"id": c.id, "type": "function",
                         "function": {"name": c.name, "arguments": json.dumps(c.arguments)}}
                        for c in m.tool_calls]
                out.append(msg)
            else:
                out.append({"role": "user", "content": m.text})
        return out

    async def complete(self, model_id, req):
        kwargs = {"model": model_id, "messages": self._messages(req),
                  self.token_param: req.max_tokens}
        if model_id not in NO_TEMPERATURE:
            kwargs["temperature"] = req.temperature
        if req.tools:
            kwargs["tools"] = [{"type": "function", "function": {
                "name": t.name, "description": t.description, "parameters": t.parameters}}
                for t in req.tools]
        if req.json_schema:
            kwargs["response_format"] = {"type": "json_schema", "json_schema": {
                "name": "result", "strict": True, "schema": self.schema_fn(req.json_schema)}}
        kwargs.update(req.options.get(self.name, {}))
        try:
            r = await self.client.chat.completions.create(**kwargs)
        except openai.APIStatusError as e:
            raise from_status(e.status_code, str(e), retry_after(e.response),
                              getattr(e, "code", None)) from e
        except (openai.APITimeoutError, openai.APIConnectionError) as e:
            raise Transient(str(e)) from e

        choice = r.choices[0]
        text = choice.message.content or ""
        calls = [ToolCall(c.id, c.function.name, as_dict(c.function.arguments))
                 for c in (choice.message.tool_calls or [])]
        refused = getattr(choice.message, "refusal", None)
        stop = "refusal" if refused else FINISH_MAP.get(choice.finish_reason, "other")
        data = json.loads(text) if req.json_schema and text and stop == "end" else None
        details = getattr(r.usage, "prompt_tokens_details", None)
        usage = Usage(r.usage.prompt_tokens, r.usage.completion_tokens,
                      getattr(details, "cached_tokens", 0) or 0)
        return Response(text, calls, stop, usage, self.name, model_id, data)

class MistralAdapter(OpenAIAdapter):
    name = "mistral"
    token_param = "max_tokens"
    schema_fn = staticmethod(closed)

    def __init__(self, api_key=None):
        super().__init__(api_key=api_key, base_url="https://api.mistral.ai/v1")

Per-model quirks live here and nowhere else. The NO_TEMPERATURE set is the seed of a capability table: as you add models, record what each one rejects, so application code never has to know.

Step 6: The Gemini adapter

Gemini is the odd one out. It names the assistant role model, matches function responses by name rather than ID, and takes its system prompt and limits in a config object. The adapter hides all of that, including inventing tool call IDs so the rest of your code stays uniform.

import uuid
from google import genai
from google.genai import errors as gerrors

FINISH = {"STOP": "end", "MAX_TOKENS": "length", "SAFETY": "refusal",
          "RECITATION": "refusal", "BLOCKLIST": "refusal", "PROHIBITED_CONTENT": "refusal"}

class GeminiAdapter:
    name = "gemini"

    def __init__(self, api_key=None):
        self.client = genai.Client(api_key=api_key)

    def _contents(self, msgs):
        names = {c.id: c.name for m in msgs for c in m.tool_calls}   # results are matched by name
        out = []
        for m in msgs:
            if m.role == "tool":
                part = {"function_response": {"name": names[m.tool_call_id],
                                              "response": {"result": m.text}}}
                if out and out[-1]["role"] == "user" and "function_response" in out[-1]["parts"][-1]:
                    out[-1]["parts"].append(part)
                else:
                    out.append({"role": "user", "parts": [part]})
            elif m.role == "assistant":
                parts = [{"text": m.text}] if m.text else []
                parts += [{"function_call": {"name": c.name, "args": c.arguments}}
                          for c in m.tool_calls]
                out.append({"role": "model", "parts": parts})
            else:
                out.append({"role": "user", "parts": [{"text": m.text}]})
        return out

    async def complete(self, model_id, req):
        config = {"max_output_tokens": req.max_tokens, "temperature": req.temperature}
        if req.system:
            config["system_instruction"] = req.system
        if req.tools:
            config["tools"] = [{"function_declarations": [
                {"name": t.name, "description": t.description,
                 "parameters": gemini_safe(t.parameters)} for t in req.tools]}]
        if req.json_schema:
            config["response_mime_type"] = "application/json"
            config["response_schema"] = gemini_safe(req.json_schema)
        config.update(req.options.get(self.name, {}))
        try:
            r = await self.client.aio.models.generate_content(
                model=model_id, contents=self._contents(req.messages), config=config)
        except gerrors.APIError as e:
            raise from_status(e.code, str(e)) from e

        cand = r.candidates[0] if r.candidates else None
        parts = cand.content.parts if cand and cand.content and cand.content.parts else []
        text = "".join(p.text for p in parts if p.text)
        calls = [ToolCall(fc.id or "call_" + uuid.uuid4().hex[:12], fc.name, dict(fc.args or {}))
                 for fc in (r.function_calls or [])]
        if cand is None:
            stop = "refusal"                      # the prompt itself was blocked
        elif calls:
            stop = "tool_use"
        else:
            stop = FINISH.get(cand.finish_reason.name if cand.finish_reason else "STOP", "other")
        data = json.loads(text) if req.json_schema and text and stop == "end" else None
        meta = r.usage_metadata
        usage = Usage(getattr(meta, "prompt_token_count", 0) or 0,
                      getattr(meta, "candidates_token_count", 0) or 0,
                      getattr(meta, "cached_content_token_count", 0) or 0)
        return Response(text, calls, stop, usage, self.name, model_id, data)

The names map is what makes cross-provider failover work. If Claude issued a tool call and Gemini answers the next turn, the history still contains a Claude-style ID, and the adapter resolves it to a function name on the way out. Your tool loop never notices.

Step 7: The client: routes, retries, breaker, fallback, cost

Routing is a dictionary of ordered routes per alias, each with its price. The client tries providers in order, retries inside a provider only for retryable errors, falls over only for fallback-worthy ones, and skips any provider whose breaker is open. The retry delay is full-jitter backoff, honoring Retry-After when it is short, as explained in the Exponential Backoff guide.

import asyncio, random, time
from dataclasses import dataclass

@dataclass
class Route:
    provider: str
    model_id: str
    in_per_m: float          # USD per million input tokens
    out_per_m: float         # USD per million output tokens

class Breaker:               # simplified: closed, open, then one probe after the cooldown
    def __init__(self, threshold=5, cooldown=30.0):
        self.threshold, self.cooldown = threshold, cooldown
        self.failures, self.opened_at = 0, 0.0

    def allow(self):
        return self.failures < self.threshold or time.monotonic() - self.opened_at > self.cooldown

    def ok(self):
        self.failures = 0

    def bad(self):
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = time.monotonic()

class LLMClient:
    def __init__(self, adapters, routes, max_attempts=3, timeout=60.0, sleep=asyncio.sleep):
        self.adapters = {a.name: a for a in adapters}
        self.routes = routes
        self.breakers = {name: Breaker() for name in self.adapters}
        self.max_attempts, self.timeout, self.sleep = max_attempts, timeout, sleep

    async def complete(self, req):
        tried, last = [], None
        for route in self.routes[req.model]:
            breaker = self.breakers[route.provider]
            if not breaker.allow():
                continue
            try:
                resp = await self._attempt(route, req)
            except LLMError as e:
                last = e
                if not e.fallback:
                    raise                      # a bad request will fail everywhere
                breaker.bad()
                tried.append(route.provider)
                continue
            breaker.ok()
            resp.fell_back_from = tried
            resp.cost_usd = (resp.usage.input_tokens * route.in_per_m
                             + resp.usage.output_tokens * route.out_per_m) / 1_000_000
            return resp
        raise last or Transient("no provider available")

    async def _attempt(self, route, req):
        adapter = self.adapters[route.provider]
        for attempt in range(1, self.max_attempts + 1):
            try:
                resp = await asyncio.wait_for(adapter.complete(route.model_id, req), self.timeout)
                resp.attempts = attempt
                return resp
            except asyncio.TimeoutError:
                err = Transient(f"{route.provider} timed out after {self.timeout}s")
            except LLMError as e:
                err = e
            if not err.retryable or attempt == self.max_attempts:
                raise err
            if err.retry_after and err.retry_after > 30:
                raise err                      # asked to wait too long: fail over instead
            delay = err.retry_after or random.uniform(0, min(30.0, 2 ** attempt))
            await self.sleep(delay)

Timeouts are enforced here, with asyncio.wait_for, instead of inside each SDK, because SDK timeout semantics differ between providers. One rule, applied once. For the surrounding theory, see the Circuit Breakers guide and the Fallback Strategies guide.

Wiring it up takes one block of configuration. Prices below come from our provider hubs, and the GPT figures are approximate, so keep real prices in a config file you can update without a deploy:

client = LLMClient(
    adapters=[AnthropicAdapter(), OpenAIAdapter(), GeminiAdapter(), MistralAdapter()],
    routes={
        "smart": [Route("anthropic", "claude-sonnet-4-6", 3.00, 15.00),
                  Route("openai", "gpt-5.6", 5.00, 20.00),          # approximate
                  Route("gemini", "gemini-3.5-pro", 3.50, 10.50)],
        "cheap": [Route("gemini", "gemini-3.5-flash", 0.30, 2.50),
                  Route("openai", "gpt-5.6-mini", 0.50, 2.00),      # approximate
                  Route("mistral", "mistral-small-latest", 0.40, 1.20)],
    },
)

resp = await client.complete(Request(
    model="smart",
    system="You are terse.",
    messages=[Message("user", "Summarize HTTP caching in one sentence.")],
))
print(resp.provider, resp.attempts, resp.fell_back_from, round(resp.cost_usd, 6))

Cost here counts input and output tokens only. Cached tokens are reported separately in usage.cached_tokens because providers price them differently, and you will want your own dashboards to apply a per-route cached rate.

An agent loop that does not care who answered

Because history is stored in your format, a tool loop is provider-neutral, and a failover in the middle of it just works:

async def run_agent(client, req, handlers, max_turns=8):
    for _ in range(max_turns):
        resp = await client.complete(req)
        if not resp.tool_calls:
            return resp
        req.messages.append(Message("assistant", resp.text, resp.tool_calls))
        for call in resp.tool_calls:
            result = await handlers[call.name](**call.arguments)
            req.messages.append(Message("tool", json.dumps(result), tool_call_id=call.id))
    raise RuntimeError("tool loop did not converge")

Validate call.arguments before executing a handler. Malformed or invented arguments are the most common agent bug, and the Function Calling Reliability guide covers the validation patterns.

Step 8: Streaming and the failover rule

Streaming adds one adapter method, stream(model_id, req), an async generator that yields normalized deltas: a text chunk, a completed tool call, or a final usage and stop event. Each adapter translates its provider's event format, including Claude's typed events and the partial tool-argument JSON that arrives in fragments and must only be parsed once the block closes.

The client's streaming method has one extra rule that the non-streaming path does not need:

    async def stream(self, req):
        emitted = False
        for route in self.routes[req.model]:
            if not self.breakers[route.provider].allow():
                continue
            try:
                async for delta in self.adapters[route.provider].stream(route.model_id, req):
                    emitted = True
                    yield delta
                return
            except LLMError as e:
                if emitted or not e.fallback:
                    raise                  # a user already saw tokens: do not splice answers
                self.breakers[route.provider].bad()
        raise Transient("no provider available")

Once the first token has reached the user, you cannot silently switch providers, because the second answer would continue a sentence the first one started. Fail over only before the first token, and otherwise surface the error and let the caller retry the whole request.

What not to abstract

The urge to unify everything is how these libraries rot. Keep the shared interface small and route the rest through options:

FeatureWhy it leaksWhat to do
Prompt cachingClaude uses cache_control markers, OpenAI caches long prefixes automatically, Gemini uses explicit cached-content objectsPer-provider options; keep stable content first in every prompt
Extended thinking and reasoning controlsDifferent parameters, budgets, and output shapesPer-provider options
Built-in web search and groundingProvider-hosted tools with different citation formatsoptions, or your own retrieval tool
Batch APIsAsynchronous with different lifecyclesA separate module, not complete()
Files, images, audio, videoDifferent upload flows, limits, and supported typesAdd a parts field later, adapter by adapter
Per-model parameter rulesReasoning models reject settings others acceptA capability table inside the adapter

Testing without spending money

The client has no SDK dependency, so it tests with a fake adapter that plays back a script of responses and errors. Inject the sleep function so tests do not wait on backoff.

class FakeAdapter:
    def __init__(self, name, script):
        self.name, self.script = name, list(script)

    async def complete(self, model_id, req):
        step = self.script.pop(0)
        if isinstance(step, Exception):
            raise step
        return step

async def no_sleep(_):
    pass

async def test_falls_back_when_primary_is_overloaded():          # pytest-asyncio
    ok = Response("hi", [], "end", Usage(10, 5), "openai", "gpt-5.6")
    client = LLMClient(
        adapters=[FakeAdapter("anthropic", [Overloaded("busy")] * 3), FakeAdapter("openai", [ok])],
        routes={"smart": [Route("anthropic", "a", 1, 1), Route("openai", "b", 1, 1)]},
        sleep=no_sleep,
    )
    resp = await client.complete(Request("smart", [Message("user", "hi")]))
    assert resp.provider == "openai"
    assert resp.fell_back_from == ["anthropic"]

The adapters need their own tests. Check the conversion code (your messages to the provider's format, the provider's response back to Response) against recorded request and response fixtures, and run a small live contract suite on a schedule rather than on every commit. The same scenarios run against every adapter:

ScenarioEvery adapter must return
Plain promptstop == "end", non-empty text, usage above zero
Tiny max_tokensstop == "length"
One tool call, then a resultA ToolCall with an ID, then a final text answer
Two parallel tool callsBoth results accepted in a single follow-up
JSON schema requestdata is a dict matching the schema
Bad API keyAuthFailed, not an SDK exception
Bad model nameBadRequest
History from another providerAccepted, with tool IDs resolved correctly

The last row is the one most clients forget, and it is the one that proves failover works.

Pitfalls to avoid

  • Stacking retry layers. SDK retries plus client retries multiplies attempts. Set max_retries=0 in every SDK, as the adapters above do.
  • Catching SDK exceptions in application code. The moment one leaks, your abstraction has a hole. Translate at the adapter boundary, always.
  • Treating a 200 as success. Refusals, truncations, and schema-invalid output all return 200. Check stop and validate data.
  • Silent provider changes in tests. A fallback that quietly answers every request hides an outage. Alert on fell_back_from being non-empty.
  • Hard-coded prices. They change. Keep them in config and review monthly.
  • Unifying too much. Every field you add to the shared types is a field four providers must honor. Prefer options until two providers genuinely need the same thing.

For the load-balancing side of this design, with health scoring and cost-aware routing across providers, see our Multi-Provider Load Balancer post. For the single-provider hardening that sits beneath each adapter, see the Bulletproof Claude API Wrapper.

Frequently asked questions

Why not just use LiteLLM or OpenRouter?

You can, and for many teams a gateway is the right call. Build your own when you need precise control over which errors retry and which fall back, when traffic cannot pass through a third party, or when you want a small surface you fully understand. If you buy, use the design rules in this post as your evaluation checklist.

Can I use the OpenAI SDK for every provider through compatible endpoints?

For quick tests, yes. Gemini and Anthropic both offer OpenAI-compatible endpoints, and Mistral's chat API is OpenAI-shaped. In production, native adapters give you access to caching, extended thinking, grounding, and accurate error semantics that compatibility layers hide or approximate.

How do I handle prompt caching in a provider-agnostic client?

Do not abstract it. Anthropic uses cache_control markers, OpenAI caches long prefixes automatically, and Gemini uses explicit cached content objects. Put provider-specific cache settings in the per-provider options dict, and keep prompts structured with stable content first so every provider's cache can hit.

How do I keep conversation history portable across providers?

Store history in your own normalized Message format, with tool calls and results as structured fields. Each adapter converts to the provider format on every request. That is what makes failover in the middle of a conversation possible.

Where do tool call IDs come from on Gemini?

Gemini matches function responses by function name and may not return a call ID. The adapter generates an ID when it reads the response and keeps an ID-to-name map when it converts history back. Your application code only ever sees IDs.

How many retries per provider before failing over?

Two or three attempts with full-jitter backoff, honoring Retry-After when it is short. If a provider asks you to wait longer than your latency budget, fail over immediately instead of sleeping.

How do I normalize token counts and cost?

Map each SDK's usage fields to input, output, and cached tokens, then multiply by a per-route price table. Keep prices in config, since they change. Tokenizers differ, so compare cost per successful task rather than raw token counts.

How do I test adapters without spending money?

Unit test the client with a fake adapter that returns scripted responses and errors. Test each real adapter's conversion code with recorded request and response fixtures. Run a small live contract suite on a schedule, not on every commit.

Can I fail over in the middle of a stream?

Only before the first token reaches the user. After that, switching providers would splice two different answers together. Surface the error, or let the caller retry the whole request.

How do I add a fifth provider?

Write one adapter that implements complete, map its exceptions through the shared status function, add routes in config, and run the contract suite against it. If the provider is OpenAI-shaped, subclass the OpenAI adapter and change the base URL, as the Mistral adapter does.

Keep reading