When you need to fallback from gpt-4o to claude because OpenAI is throwing 429s or hard timeouts, a naive try/except around one SDK call isn’t enough. You need to preserve the conversation, map message formats, and classify which errors are worth retrying on the secondary provider. This pattern is boring infrastructure work, but it’s the difference between a demo and a system that survives a provider incident.
Step 1: Define a provider-agnostic request shape
Don’t pass OpenAI-specific objects deep into your call logic. Define a minimal request contract that both providers can consume after translation. At minimum you need the message list, a max token budget, and temperature.
from dataclasses import dataclass, field
from typing import Literal
@dataclass
class ChatRequest:
messages: list[dict] # [{"role": "user", "content": "..."}]
max_tokens: int = 1024
temperature: float = 0.7
primary_model: str = "gpt-4o"
fallback_model: str = "claude-3-5-sonnet-20240620"Keep the message format in OpenAI’s shape (role/content) as your internal canonical representation. Anthropic splits system from messages and requires max_tokens, so you’ll translate at the edge.
Step 2: Implement a sequential caller with error classification
The core of any fallback from gpt-4o to claude is a function that tries the primary, catches retryable transport/auth errors, then calls the secondary. Only retryable errors should trigger the fallback; a 400 from malformed input should bubble up immediately.
import openai
import anthropic
from openai import RateLimitError, APIConnectionError, APITimeoutError
from anthropic import RateLimitError as AnthropicRateLimit, APIConnectionError as AnthropicConnError
def chat_with_fallback(req: ChatRequest) -> str:
try:
return _call_openai(req)
except (RateLimitError, APIConnectionError, APITimeoutError) as e:
# Retryable on OpenAI side -> fall back
print(f"gpt-4o failed: {e}; falling back to claude")
try:
return _call_anthropic(req)
except (AnthropicRateLimit, AnthropicConnError) as e2:
raise RuntimeError("Both providers failed") from e2
except Exception as e:
# Non-retryable (bad request, auth) -> do not fall back
raiseThis is the minimal skeleton. In production you’ll want to log structured events and probably emit metrics on fallback rate.
Step 3: Normalize responses from both APIs
OpenAI returns choices[0].message.content. Anthropic returns a content list of blocks where type == "text". Write a thin normalization layer so the rest of your app never knows which model answered.
def _call_openai(req: ChatRequest) -> str:
client = openai.OpenAI() # reads OPENAI_API_KEY
resp = client.chat.completions.create(
model=req.primary_model,
messages=req.messages,
max_tokens=req.max_tokens,
temperature=req.temperature,
)
return resp.choices[0].message.content
def _call_anthropic(req: ChatRequest) -> str:
client = anthropic.Anthropic() # reads ANTHROPIC_API_KEY
# Extract system prompt if present
system = ""
msgs = []
for m in req.messages:
if m["role"] == "system":
system = m["content"]
else:
msgs.append(m)
resp = client.messages.create(
model=req.fallback_model,
system=system,
messages=msgs,
max_tokens=req.max_tokens,
temperature=req.temperature,
)
return "".join(b.text for b in resp.content if b.type == "text")Note that Anthropic requires max_tokens on every call; OpenAI treats it as optional but you should set it anyway to bound cost.
Step 4: Preserve conversation state across the fallback
A fallback from gpt-4o to claude must not lose the prior turns. If you’re building a chat loop, append the normalized assistant reply to your message list regardless of source, then continue. The user shouldn’t see a truncated context.
def run_conversation(initial: list[dict]):
req = ChatRequest(messages=initial)
assistant_text = chat_with_fallback(req)
req.messages.append({"role": "assistant", "content": assistant_text})
# next user turn uses req.messages as-is
return req.messagesIf the primary fails mid-stream (you’re using streaming), you cannot resume the same SSE connection on the fallback. Buffer the streamed output and only commit to history after the call fully completes; on failure, discard the partial buffer and call the fallback non-streaming or with its own stream.
Step 5: Add timeouts and a circuit breaker
Provider degradation often manifests as slow hangs, not errors. Set explicit client timeouts. Both SDKs accept a timeout param (or an httpx.Client underneath).
openai_client = openai.OpenAI(timeout=10.0)
anthropic_client = anthropic.Anthropic(timeout=10.0)For higher traffic, add a circuit breaker that stops calling a provider after N consecutive failures within a window. A simple in-memory decorator works for single-instance services; use Redis for distributed. Without this, a dead provider will add 10s latency to every request before falling back.
from functools import wraps
import time
def circuit_breaker(failures=5, cooldown=30):
state = {"f": 0, "open_until": 0.0}
def deco(fn):
@wraps(fn)
def wrapped(*a, **k):
if time.time() < state["open_until"]:
raise RuntimeError("circuit open")
try:
r = fn(*a, **k)
state["f"] = 0
return r
except (RateLimitError, APIConnectionError, APITimeoutError):
state["f"] += 1
if state["f"] >= failures:
state["open_until"] = time.time() + cooldown
raise
return wrapped
return decoStep 6: Verify the fallback works
You can’t claim the fallback from gpt-4o to claude is working without forcing the primary to fail in a test. Use a bad API key or a proxy that returns 429. Below is a pytest sketch using respx or simple env toggling.
import os
import pytest
def test_fallback_triggers(monkeypatch):
# Force OpenAI client to raise RateLimitError
import openai
def fake_create(*args, **kwargs):
raise openai.RateLimitError("rate", response=None, body=None)
monkeypatch.setattr(openai.OpenAI, "chat", type("C", (), {"completions": type("Comp", (), {"create": staticmethod(fake_create)})}) )
# Set a dummy Anthropic key and mock its success
monkeypatch.setenv("ANTHROPIC_API_KEY", "test")
import anthropic
def fake_anthropic(*a, **k):
class Block: type="text"; text="fallback ok"
class Resp: content=[Block()]
return Resp()
monkeypatch.setattr(anthropic.Anthropic, "messages", type("M", (), {"create": staticmethod(fake_anthropic)}) )
from mymodule import ChatRequest, chat_with_fallback
r = chat_with_fallback(ChatRequest(messages=[{"role":"user","content":"hi"}]))
assert r == "fallback ok"Run it with:
pytest tests/test_fallback.py -qAdditionally, run a live smoke test in staging with OPENAI_API_KEY unset to confirm the Claude path returns coherent text. Watch logs for the “gpt-4o failed” line so you know the branch executed.
Step 7: Offload fallback to a gateway when it makes sense
Hand-rolled fallback is fine until you support ten models and three regions. An OpenAI-compatible inference gateway such as n4n.ai performs automatic fallback when a provider is rate-limited or degraded, and honors client routing directives, so you can send one request and let the gateway shift from gpt-4o to claude-3-5-sonnet without client code. If you already standardize on the OpenAI request shape, point your base_url at the gateway and drop the custom caller. For most teams past prototype stage, that’s the lower-maintenance path.
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key="your-key")
# Gateway routes to gpt-4o, falls back to claude on provider errors
resp = client.chat.completions.create(model="gpt-4o", messages=[{"role":"user","content":"hi"}])Either approach is valid. The manual pattern gives you precise control over error taxonomy and context mapping; the gateway pattern removes the maintenance tax. Pick based on how many providers you actually run today, not on hype.