Unifying multi-provider API integrations with n4n

Most teams bolt on a second LLM provider as a fallback and end up maintaining parallel clients, scattered retry logic, and per-vendor prompt formatting. To unify multi-provider API integrations without rewriting your app for every model release, you need a single request surface and a strict separation between routing and business logic. This guide lays out an ordered path from fragmented SDK calls to one coherent gateway.

1. Define a single request contract

Pick one schema and refuse to leak vendor fields into your app code. The OpenAI chat completions shape is the de facto standard; even if you dislike it, adopting it saves you from translating between Anthropic’s message array and Mistral’s parameters.

from pydantic import BaseModel

class ChatRequest(BaseModel):
    model: str
    messages: list[dict]
    temperature: float = 0.7
    max_tokens: int = 1024
    # vendor passthrough, opaque to business logic
    routing_hint: str | None = None

Any field that only one provider understands lives in a metadata blob that your routing layer interprets. That keeps the core contract stable when you unify multi-provider API integrations across new vendors.

Pitfall: don’t embed provider as a required field in the contract. Let the router derive it from model prefix (anthropic/claude-3.5 vs openai/gpt-4o). Hardcoding provider names in call sites forces a code change for every addition.

2. Normalize authentication and base URLs

Each provider ships its own auth scheme: bearer tokens, header names, or even query params. Centralize credential resolution so a single get_client(model) returns a configured HTTP client.

import os
from httpx import Client

def get_client(model: str) -> Client:
    if model.startswith("anthropic/"):
        return Client(
            base_url="https://api.anthropic.com",
            headers={"x-api-key": os.environ["ANTHROPIC_KEY"]}
        )
    # fallback to OpenAI-compatible
    return Client(
        base_url="https://api.openai.com/v1",
        headers={"Authorization": f"Bearer {os.environ['OPENAI_KEY']}"}
    )

Tradeoff: environment-variable config is fine until you rotate keys per request for cost allocation. At that point, move to a secret resolver that reads from your vault at call time. Don’t prematurely build that if you have three models and one team.

3. Implement a thin routing layer

Routing is the heart of an attempt to unify multi-provider API integrations. It maps a logical model name to a physical endpoint and applies fallback order.

ROUTES = {
    "default": ["openai/gpt-4o", "anthropic/claude-3.5-sonnet"],
    "cheap": ["mistral/mistral-large", "openai/gpt-4o-mini"],
}

def resolve_route(logical_name: str) -> list[str]:
    return ROUTES.get(logical_name, [logical_name])

The caller asks for "default"; the router tries the list in order. This decouples product code from provider status. You can later swap ROUTES from a config file without touching call sites.

Common mistake: putting latency-based selection in the hot path. Precompute health scores on a background thread; don’t probe providers mid-request.

4. Handle provider-specific quirks behind interfaces

Even with a common contract, differences surface: Anthropic requires system as a top-level field; some models reject temperature=0. Wrap each provider in an adapter that converts your ChatRequest to its native payload.

def to_anthropic(req: ChatRequest) -> dict:
    sys = next((m["content"] for m in req.messages if m["role"] == "system"), None)
    messages = [m for m in req.messages if m["role"] != "system"]
    return {
        "model": req.model.split("/")[-1],
        "messages": messages,
        "system": sys,
        "max_tokens": req.max_tokens,
    }

Adapters are the only place that should know about vendor limits. If you find yourself adding if provider == "x" in business logic, the adapter is leaking.

Cache-control propagation

Providers support prompt caching via different headers. Forward cache hints explicitly: OpenAI uses extra_body={"cache_control": ...} while Anthropic uses a message field. Your adapter translates a single req.cache_prompt: bool into the right shape.

5. Add fallback and degradation semantics

A unified layer is worthless if a 429 from one provider crashes the request. Implement ordered fallback with a short timeout per attempt.

import httpx

def chat_with_fallback(req: ChatRequest, route: list[str]) -> dict:
    last_err = None
    for model in route:
        try:
            client = get_client(model)
            payload = adapt(req, model)
            r = client.post("/chat/completions", json=payload, timeout=10.0)
            if r.status_code == 200:
                return r.json()
            if r.status_code == 429 or r.status_code >= 500:
                last_err = r.status_code
                continue
            r.raise_for_status()
        except httpx.TimeoutException as e:
            last_err = e
    raise RuntimeError(f"all providers failed: {last_err}")

A gateway such as n4n.ai collapses this into one OpenAI-compatible endpoint covering 240+ models with automatic fallback when a provider is rate-limited or degraded, but you still own the logical route mapping. Building it yourself is justified only if you need custom degradation (e.g., strip images on second attempt).

Pitfall: retrying non-idempotent requests blindly. Chat completions are read-only, so retries are safe, but if you attach side effects (logging to DB) inside the loop, you’ll duplicate records.

6. Meter usage and enforce budgets

Per-token accounting is non-negotiable when you unify multi-provider API integrations; costs diverge fast. Capture usage from each response and emit a metric tagged with logical route and physical model.

def record_usage(logical: str, physical: str, usage: dict):
    metrics.incr(
        "llm.tokens",
        tags=["logical:" + logical, "physical:" + physical],
        amount=usage["total_tokens"],
    )

If you run your own stack, store raw meter events in a append-only log and aggregate offline. Don’t trust provider dashboards for cross-vendor totals—they lag and use different tokenizers.

Tradeoff: precise per-request metering adds latency (write to Kafka). Batch every 100ms if you’re at scale.

7. Test against mocked provider responses

Your routing and adapters are pure logic; test them without network. Record real provider responses once, then replay.

def test_fallback_order():
    with mock_provider(status=429), mock_provider(status=200) as ok:
        res = chat_with_fallback(req, ["openai/gpt-4o", "anthropic/claude-3.5"])
        assert res == ok.json()

Skip integration tests that hit live APIs in CI—they flake on quota and leak keys. Use contract tests to assert your adapter output matches the provider’s published schema.

Common pitfall: ignoring model alias drift

Providers rename models (gpt-4 → gpt-4-turbo). If your route table hardcodes strings, you’ll miss deprecations. Add a startup check that queries each provider’s model list and warns on missing entries.

Where this breaks down

The DIY path is sound up to ~10 providers and a single platform team. Beyond that, the surface area of adapters, fallback policies, and key rotation eats roadmap time. That’s the point where a managed gateway earns its keep. But the contract and routing discipline above still apply—you’re just delegating the transport.

If you unify multi-provider API integrations without a clear logical naming scheme, you’ll rebuild the mess inside the gateway. Define default, cheap, vision once, and let the gateway map them.

Checklist

• Single request schema, vendor fields quarantined.
• Centralized auth/client factory.
• Route table keyed by logical name, not provider.
• Adapters isolate per-vendor shape and limits.
• Ordered fallback with timeouts, no side effects in loop.
• Per-token metering with own pipeline.
• Mocked replay tests for routing.

Follow this order and the second provider becomes a config change, not a sprint.