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.
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
- Normalize to the smallest useful core. Text, tool calls, JSON output, usage, and a stop reason. Nothing else belongs in the shared types.
- Errors are part of the interface. Callers catch your exception classes and never an SDK's.
- 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.
- Leak the other 20 percent on purpose. Caching, extended thinking, and grounding differ too much to unify. Give them a per-provider
optionsescape hatch instead of pretending. - Routing is data. Application code asks for
"smart"or"cheap". Config decides which providers and models answer, and in what order. - 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:
| Claude | GPT | Gemini | Mistral | |
|---|---|---|---|---|
| System prompt | top-level system | first message, role system | system_instruction in config | first message, role system |
| Assistant role name | assistant | assistant | model | assistant |
| Tool arguments arrive as | parsed object | JSON string | parsed object | JSON string |
| Tool result matched by | tool_use_id | tool_call_id | function name | tool_call_id |
| Truncation signal | max_tokens | length | MAX_TOKENS | length |
| Output cap parameter | max_tokens (required) | max_completion_tokens | max_output_tokens | max_tokens |
| Overload status | 529 | 503 | 503 | 503 |
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:
| Feature | Why it leaks | What to do |
|---|---|---|
| Prompt caching | Claude uses cache_control markers, OpenAI caches long prefixes automatically, Gemini uses explicit cached-content objects | Per-provider options; keep stable content first in every prompt |
| Extended thinking and reasoning controls | Different parameters, budgets, and output shapes | Per-provider options |
| Built-in web search and grounding | Provider-hosted tools with different citation formats | options, or your own retrieval tool |
| Batch APIs | Asynchronous with different lifecycles | A separate module, not complete() |
| Files, images, audio, video | Different upload flows, limits, and supported types | Add a parts field later, adapter by adapter |
| Per-model parameter rules | Reasoning models reject settings others accept | A 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:
| Scenario | Every adapter must return |
|---|---|
| Plain prompt | stop == "end", non-empty text, usage above zero |
Tiny max_tokens | stop == "length" |
| One tool call, then a result | A ToolCall with an ID, then a final text answer |
| Two parallel tool calls | Both results accepted in a single follow-up |
| JSON schema request | data is a dict matching the schema |
| Bad API key | AuthFailed, not an SDK exception |
| Bad model name | BadRequest |
| History from another provider | Accepted, 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=0in 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
stopand validatedata. - Silent provider changes in tests. A fallback that quietly answers every request hides an outage. Alert on
fell_back_frombeing 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
optionsuntil 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.