Migrating from OpenAI SDK to a gateway with zero downtime

When you need to move production traffic from direct OpenAI calls to a unified gateway, the safest path is a staged cutover that keeps the old client running until the new one is proven. A zero downtime openai sdk migration is less about rewriting business logic and more about isolating the client dependency and routing traffic incrementally behind a flag.

Step 1: Inventory every OpenAI SDK call site

Before changing anything, find all direct usages. In a Python service, grep for the client construction and the completion calls:

grep -rn "openai\." --include="*.py" . | head -50
grep -rn "ChatCompletion\|chat.completions.create" --include="*.py" .

Record the models, temperature, max_tokens, and any retry/timeout configuration. If you use the Node SDK, do the same with createChatCompletion or chat.completions.create. The goal is a complete map of request shapes so the gateway receives identical payloads.

Pay attention to streaming vs non-streaming, and to any use of response_format or function tools. Gateways that mirror the OpenAI schema pass these through, but you must confirm. Also note custom timeout or max_retries passed to the OpenAI constructor; the gateway may have different default limits, and you want parity.

A practical inventory output is a JSON file you can reference later:

{
  "call_sites": [
    {
      "file": "src/summarize.py",
      "model": "gpt-4o-mini",
      "stream": false,
      "timeout": 30,
      "max_retries": 2
    }
  ]
}

Step 2: Wrap the SDK in a thin client interface

Don’t scatter OpenAI() constructors across the codebase. Define a protocol that matches the subset you use:

from typing import Protocol, Any
from openai import OpenAI

class ChatClient(Protocol):
    def complete(self, model: str, messages: list[dict], **kwargs: Any) -> Any:
        ...

class DirectOpenAIClient:
    def __init__(self, api_key: str, timeout: int = 30) -> None:
        self._client = OpenAI(api_key=api_key, timeout=timeout, max_retries=2)

    def complete(self, model: str, messages: list[dict], **kwargs: Any) -> Any:
        return self._client.chat.completions.create(
            model=model, messages=messages, **kwargs
        )

This takes an afternoon and immediately decouples your code from the concrete SDK. All call sites now use ChatClient.complete. That is the seam you will exploit for the zero downtime openai sdk migration. If you use async, define a second method acomplete backed by AsyncOpenAI; the gateway swap is identical.

Step 3: Point the wrapper at a gateway endpoint

Most gateways speak the OpenAI HTTP protocol. You only need to change base_url and keep the same request objects. If you use n4n.ai, its single OpenAI-compatible endpoint fronts 240+ models and applies automatic fallback when a provider is degraded, but the client code is identical to DirectOpenAIClient except for the URL and key.

class GatewayClient:
    def __init__(self, base_url: str, api_key: str, timeout: int = 30) -> None:
        self._client = OpenAI(base_url=base_url, api_key=api_key, timeout=timeout, max_retries=2)

    def complete(self, model: str, messages: list[dict], **kwargs: Any) -> Any:
        # Gateway forwards cache-control hints and honors routing directives
        return self._client.chat.completions.create(
            model=model, messages=messages, **kwargs
        )

For TypeScript services, the same pattern holds:

import OpenAI from "openai";

const client = new OpenAI({
  baseURL: "https://gateway.example.com/v1",
  apiKey: process.env.GATEWAY_KEY,
  timeout: 30,
  maxRetries: 2,
});

No request schema changes. That is the entire migration surface. If you previously set openai.api_key globally, stop; explicit clients avoid ambient state during the dual-run period.

Step 4: Shadow traffic and diff responses

Run both clients in parallel, sending production requests to the direct client and copying them to the gateway. Compare outputs to catch schema or behavior drift before users see it.

import logging
from typing import Any

def shadow_complete(direct: ChatClient, gateway: ChatClient,
                     model: str, messages: list[dict], **kwargs: Any) -> Any:
    result = direct.complete(model, messages, **kwargs)
    try:
        gw_result = gateway.complete(model, messages, **kwargs)
        if gw_result.choices[0].message.content != result.choices[0].message.content:
            logging.warning("Gateway response mismatch for model %s", model)
    except Exception as e:
        logging.error("Gateway shadow call failed: %s", e)
    return result

Keep the direct result as the one returned to the user. The gateway call is fire-and-forget for observation. Run this for a few hours across peak load. If mismatch rate is non-zero but acceptable (e.g., non-deterministic sampling), note it; if the gateway throws validation errors, fix the request shape. For streaming endpoints, compare token counts rather than exact strings.

Step 5: Roll out with a feature flag

Cutover should be percentage-based, not binary. Use an environment variable or a real flag system:

import os
from random import random

def get_client() -> ChatClient:
    if os.getenv("USE_GATEWAY") == "1" and random() < float(os.getenv("GATEWAY_PCT", "0.0")):
        return gateway_client
    return direct_client

Start at 1%, watch error rates and latency. Bump to 10%, 50%, then 100%. Because the interface is identical, a bad gateway response only affects the flagged slice, and you can flip back instantly by setting GATEWAY_PCT=0. A zero downtime openai sdk migration relies on this kill switch; do not skip it.

If your gateway supports per-token usage metering, export those metrics to your existing dashboard. They replace the direct OpenAI billing export and let you spot cost regressions per route.

Step 6: Verify success in production

Verification is not “it didn’t 500.” You need three signals:

1. Error rate parity – gateway 5xx and timeout rate within noise of direct.
2. Latency p95 – within 10% of baseline; gateways add at most one TLS termination hop.
3. Token accounting – if the gateway returns usage objects, reconcile them with your old logging.

A minimal verification script hits both endpoints with a fixed prompt:

def verify(base_url: str, api_key: str, model: str) -> bool:
    c = OpenAI(base_url=base_url, api_key=api_key)
    resp = c.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": "ping"}],
        max_tokens=5,
    )
    return resp.choices[0].message.content is not None

Run this from a cron job against the gateway and the direct API. Alert if either fails. For streaming, assert that you receive the usage chunk at the end. Only when all three signals hold for a full day should you consider the cutover stable.

Step 7: Decommission the direct client

Once 100% traffic has flowed through the gateway for a week with clean metrics, delete DirectOpenAIClient and the flag. Your ChatClient now has one implementation. The zero downtime openai sdk migration is complete, and you have a single endpoint to rotate keys, add models, or introduce fallback without touching call sites again.

Cleanup checklist

• Remove OPENAI_API_KEY from production env if unused.
• Delete shadow logging code.
• Update onboarding docs to reference the gateway base URL.
• Confirm streaming works end-to-end with the same stream=True parameter.
• Rotate gateway keys on a schedule; the gateway abstracts provider key exposure.

Verification note

Success means: no customer-facing errors during cutover, response content equivalent for deterministic prompts, and token counts logged per request via the gateway. Run the shadow diff for at least one full traffic cycle (24h) before ramping past 10%. If you skip the shadow step, you trade a weekend outage for an afternoon of logging. The feature flag is your seatbelt—keep it until the direct client is deleted.