How to migrate 100+ OpenAI SDK calls to a gateway safely

To migrate large codebase openai sdk to gateway without breaking production, you need a mechanical strategy, not a heroic rewrite. This guide walks through a staged approach that keeps 100+ call sites functional while you swap the backend from OpenAI’s API to an OpenAI-compatible inference gateway. The steps below assume Python and the official openai package, but the pattern translates directly to TypeScript or Go.

Step 1: Audit every OpenAI SDK instantiation and usage pattern

A naive grep -r "OpenAI(" misses async clients, imported aliases, and subclassed wrappers. Start with a full AST scan that captures both OpenAI() and AsyncOpenAI() constructions, plus the method families you use (chat.completions, embeddings, audio).

grep -rn "OpenAI\|AsyncOpenAI" --include="*.py" . | wc -l

For precision, run a script that records each constructor and each create call:

import ast, pathlib

sites = []
for p in pathlib.Path(".").rglob("*.py"):
    tree = ast.parse(p.read_text(), p.name)
    for node in ast.walk(tree):
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
            if node.func.id in ("OpenAI", "AsyncOpenAI"):
                sites.append((p.name, node.lineno, node.func.id))
        if isinstance(node, ast.Attribute) and node.attr == "create":
            sites.append((p.name, node.lineno, "create"))
print(f"Found {len(sites)} relevant nodes")

Separate the inventory into three buckets: chat completions, embeddings, and streaming calls. Gateways often differ in streaming chunk shapes or embedding normalization, so you must know which code paths exercise them.

Step 2: Centralize client creation for sync and async

Do not edit 100 call sites to change base_url. Create one factory module and redirect all imports to it. Support both sync and async clients because a large codebase usually has both.

# llm_client.py
import os
from openai import OpenAI, AsyncOpenAI

def get_client() -> OpenAI:
    return OpenAI(
        base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
        api_key=os.environ.get("LLM_API_KEY", os.environ["OPENAI_API_KEY"]),
        max_retries=3,
        timeout=30,
    )

def get_async_client() -> AsyncOpenAI:
    return AsyncOpenAI(
        base_url=os.environ.get("LLM_BASE_URL", "https://api.openai.com/v1"),
        api_key=os.environ.get("LLM_API_KEY", os.environ["OPENAI_API_KEY"]),
        max_retries=3,
        timeout=30,
    )

Use a codemod or sed to replace OpenAI() → get_client() and AsyncOpenAI() → get_async_client(). After the sweep, re-run the AST script: zero constructor calls should exist outside llm_client.py.

Step 3: Drive configuration from environment

The gateway swap becomes a deploy-time decision. Set LLM_BASE_URL to your gateway’s OpenAI-compatible endpoint; leave it unset in local dev to hit OpenAI directly.

# production env
export LLM_BASE_URL="https://gateway.example.com/v1"
export LLM_API_KEY="sk-gateway-..."

A gateway like n4n.ai exposes a single OpenAI-compatible endpoint covering 240+ models and automatically falls back when a provider is rate-limited, which removes the need for client-side retry storms. Your code stays identical; only the environment changes.

Store the key in your secret manager, not in code. Rotate by updating the env var and restarting—the factory picks it up.

Step 4: Normalize model identifiers and capabilities

Gateway model names rarely match OpenAI’s gpt-4o. Centralize a mapping so call sites keep using familiar aliases.

# model_map.py
MODEL_ALIASES = {
    "gpt-4o": "openai/gpt-4o",
    "claude-3-5-sonnet": "anthropic/claude-3.5-sonnet",
    "embedding-3": "openai/text-embedding-3-small",
}

def resolve_model(name: str) -> str:
    return MODEL_ALIASES.get(name, name)

If a call site passes an unknown name, the gateway will reject it. Fail fast in tests:

def complete(messages, model="gpt-4o", **kw):
    client = get_client()
    return client.chat.completions.create(
        model=resolve_model(model), messages=messages, **kw)

For embeddings, verify dimension compatibility. Gateways may route to a different provider whose embedding size differs; assert the returned vector length in a smoke test.

Step 5: Forward routing, cache, and billing headers

Gateways honor client routing directives through headers. The OpenAI SDK accepts default_headers at construction and per-request extra_headers. Use them to pin a provider or enable prompt caching.

def get_client() -> OpenAI:
    return OpenAI(
        base_url=os.environ["LLM_BASE_URL"],
        api_key=os.environ["LLM_API_KEY"],
        default_headers={
            "X-Route": "auto",                 # gateway picks healthy provider
            "Cache-Control": "max-age=3600",   # forward cache hint
        },
    )

n4n.ai honors client routing directives and forwards provider cache-control hints, so the same header pattern works without custom middleware. For a one-off override, pass extra_headers={"X-Route": "anthropic/claude-3.5-sonnet"} to the create call.

Step 6: Handle streaming and partial responses

Streaming is where gateways diverge most. Wrap the stream so call sites receive the same Stream or AsyncStream object:

def stream_complete(messages, model="gpt-4o", **kw):
    client = get_client()
    return client.chat.completions.create(
        model=resolve_model(model),
        messages=messages,
        stream=True,
        **kw,
    )

# usage
for chunk in stream_complete([{"role": "user", "content": "hi"}]):
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

Test that the gateway’s SSE format parses with the OpenAI SDK unchanged. If the gateway adds vendor-specific fields, ignore them—the SDK tolerates unknown keys.

Step 7: Run shadow traffic and differential tests

Before cutover, send a copy of live traffic to the gateway and compare. Wrap the primary call and fire a shadow:

def shadow_complete(messages, model, **kw):
    primary = get_client().chat.completions.create(
        model=resolve_model(model), messages=messages, **kw)
    try:
        shadow_client = get_client().with_options(
            base_url=os.environ["SHADOW_URL"])
        shadow = shadow_client.chat.completions.create(
            model=resolve_model(model), messages=messages, **kw)
        # compare finish_reason, token counts, latency
        if shadow.usage.total_tokens != primary.usage.total_tokens:
            metrics.inc("shadow_token_diff")
    except Exception as e:
        metrics.inc("shadow_error", e)
    return primary

Run this for a full traffic cycle. Track divergence rate. If the gateway’s outputs differ only in latency or minor token counts, you are safe to proceed.

Step 8: Canary deployment and rollback plan

Flip one low-risk service to the gateway via env var. Keep a feature flag to revert instantly.

# service-a canary
LLM_BASE_URL="https://gateway.example.com/v1" ./run_service_a

Monitor error rates, p99 latency, and gateway health endpoints. If the gateway degrades, unset the var and restart—the factory falls back to OpenAI because the default base URL is still the original API.

Step 9: Verify success end to end

Verification is concrete. Run this checklist:

grep -rn "OpenAI(" returns only llm_client.py.
A pytest integration test hits the gateway and asserts a 200 with non-empty choices:

def test_gateway_chat():
    resp = get_client().chat.completions.create(
        model=resolve_model("gpt-4o"),
        messages=[{"role": "user", "content": "ping"}],
    )
    assert resp.choices[0].message.content

Query the gateway’s /v1/models and confirm your model is listed:

curl -H "Authorization: Bearer $LLM_API_KEY" $LLM_BASE_URL/models | jq '.data[].id'

Per-token usage metering in the gateway dashboard matches expected prompt sizes.
Production network logs show zero outbound connections to api.openai.com.

Step 10: Remove shims and document the seam

After all services run on the gateway for a week, delete shadow code and any resolve_model fallback to native OpenAI names if the gateway is now the only backend. Update the README:

All LLM calls go through llm_client.get_client() / get_async_client().
Set LLM_BASE_URL to the gateway. Model aliases live in model_map.py.
Routing headers are set in the factory.

Keep the factory. It is the seam that let you migrate large codebase openai sdk to gateway today and will let you swap gateways tomorrow without touching business logic.

Final note on safety

The migration is done when every call flows through the single client, the base URL is the gateway, and your on-call has not paged you about missing completions. Mechanical wrapping, environment-driven config, and shadow testing turn a 100+ call site nightmare into a routine deploy.