⬡ API & SDK/2026-04-11Advanced

Migrating from OpenAI to the Claude API: Code Conversion to Zero-Downtime Production Rollout (2026)

How to migrate from OpenAI GPT-4 to the Claude API: authentication, message-format conversion, streaming, tool use, error handling, and a zero-downtime phased rollout, all with full implementation code.

Claude API¹²⁸OpenAIAPI migrationGPT-4Python²² Node.js⁴ TypeScript²⁴

✦ Premium Article

Why Claude API Migrations Are Accelerating in 2026

In 2026, production teams worldwide are moving from OpenAI GPT-4 to the Claude API. The drivers are clear: Claude Sonnet 4.6 and Opus 4.6 deliver top-tier performance, the 200,000-token context window unlocks new use cases, and the pricing model is increasingly competitive.

Yet every developer tackling this migration hits the same wall: where do you even start? The two APIs share a similar philosophy but differ in enough specifics—message structure, streaming events, tool definitions, required parameters—that a naive find-and-replace breaks things fast.

Before diving in, it's worth familiarizing yourself with Claude API cost management. Claude API Cost Optimization Guide covers prompt caching and model selection strategies that will complement your migration work.

1. Understanding the Core Architectural Differences

Before writing a single line of code, internalize these structural differences. Misunderstanding them is the root cause of most migration bugs.

Message Structure

OpenAI places the system prompt inside the messages array as role: "system". Claude uses a dedicated top-level system parameter instead.

# OpenAI style
openai_request = {
    "model": "gpt-4o",
    "messages": [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain Python decorators."}
    ]
}
 
# Claude style — system is a separate top-level parameter
claude_request = {
    "model": "claude-sonnet-4-6",
    "system": "You are a helpful assistant.",  # ← top-level, not in messages
    "messages": [
        {"role": "user", "content": "Explain Python decorators."}
    ],
    "max_tokens": 1024  # ← required in Claude, optional in OpenAI
}

The Required `max_tokens` Parameter

Claude API treats max_tokens as required. Omitting it throws a validation error. In OpenAI, it's optional and defaults to model-specific limits.

Response Structure

# OpenAI response access
text = response.choices[0].message.content
 
# Claude response access
text = response.content[0].text
# response.content is a list — it can contain text blocks, tool use blocks, etc.

The list-based content structure reflects Claude's ability to interleave text, tool calls, and image outputs in a single response.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Convert OpenAI Function Calling to Claude Tool Use with working code, including tool_result block structure and stop_reason handling.

✦Prevent truncated output and instruction drift from missing max_tokens and merged system prompts, using per-task limits and system isolation.

✦Reproduce a zero-downtime phased rollout via an AIGateway, measuring cost and latency while keeping a clean rollback path.

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

2. Authentication and Client Initialization

Python (SDK)

# Before: OpenAI
from openai import OpenAI
client = OpenAI(api_key="YOUR_OPENAI_API_KEY")
 
# After: Claude (Anthropic SDK)
import anthropic
client = anthropic.Anthropic(api_key="YOUR_ANTHROPIC_API_KEY")
# Or rely on the ANTHROPIC_API_KEY environment variable:
client = anthropic.Anthropic()

Node.js / TypeScript (SDK)

// Before: OpenAI
import OpenAI from 'openai';
const openai = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });
 
// After: Claude
import Anthropic from '@anthropic-ai/sdk';
const anthropic = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY });
// Or: reads ANTHROPIC_API_KEY automatically
const anthropic = new Anthropic();

Environment Variable Setup

# .env
# Old
OPENAI_API_KEY=sk-...
 
# New (keep the old key during phased rollout)
ANTHROPIC_API_KEY=sk-ant-...
# OPENAI_API_KEY=sk-...  # Keep as fallback during migration

3. Basic Text Generation Conversion

Python (Synchronous)

import anthropic
from typing import Optional
 
def chat_with_claude(
    user_message: str,
    system_prompt: Optional[str] = None,
    model: str = "claude-sonnet-4-6",
    max_tokens: int = 2048
) -> str:
    """Drop-in replacement wrapper for OpenAI chat calls."""
    client = anthropic.Anthropic()
 
    params = {
        "model": model,
        "max_tokens": max_tokens,
        "messages": [{"role": "user", "content": user_message}]
    }
 
    # Claude rejects an empty string for system — only set it when non-empty
    if system_prompt:
        params["system"] = system_prompt
 
    response = client.messages.create(**params)
    return response.content[0].text
 
# Usage
result = chat_with_claude(
    user_message="What is asyncio in Python?",
    system_prompt="You are an experienced Python developer. Be concise."
)
print(result)

Python (Asynchronous)

import asyncio
import anthropic
from typing import Optional
 
async def async_chat_with_claude(
    user_message: str,
    system_prompt: Optional[str] = None
) -> str:
    async with anthropic.AsyncAnthropic() as client:
        params = {
            "model": "claude-sonnet-4-6",
            "max_tokens": 2048,
            "messages": [{"role": "user", "content": user_message}]
        }
        if system_prompt:
            params["system"] = system_prompt
 
        response = await client.messages.create(**params)
        return response.content[0].text
 
async def main():
    result = await async_chat_with_claude("What is Python?")
    print(result)
 
asyncio.run(main())

Multi-Turn Conversations

Multi-turn conversations follow the same messages array pattern as OpenAI, with one important rule: conversations must start and end with a user message.

def multi_turn_chat():
    client = anthropic.Anthropic()
    history = []
 
    def chat(user_input: str) -> str:
        history.append({"role": "user", "content": user_input})
 
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system="You are a helpful coding assistant.",
            messages=history
        )
 
        assistant_reply = response.content[0].text
        history.append({"role": "assistant", "content": assistant_reply})
        return assistant_reply
 
    print(chat("What is FastAPI?"))
    print(chat("What are its three main advantages?"))
    print(chat("Show me a minimal Hello World example."))
 
multi_turn_chat()

4. Streaming Response Migration

Streaming is one of the biggest implementation differences. Claude uses different event types and a context-manager pattern that feels cleaner than OpenAI's iterator-based approach.

Python Streaming

import anthropic
 
def stream_chat(user_message: str, system_prompt: str = "") -> str:
    client = anthropic.Anthropic()
    full_response = ""
 
    params = {
        "model": "claude-sonnet-4-6",
        "max_tokens": 2048,
        "messages": [{"role": "user", "content": user_message}]
    }
    if system_prompt:
        params["system"] = system_prompt
 
    # Recommended: use the context-manager streaming interface
    with client.messages.stream(**params) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
            full_response += text
 
    print()  # newline after streaming finishes
    return full_response
 
stream_chat(
    "Explain Python decorators in detail.",
    "Provide clear technical explanations."
)

Node.js Streaming

import Anthropic from '@anthropic-ai/sdk';
 
const anthropic = new Anthropic();
 
async function streamChat(
  userMessage: string,
  systemPrompt?: string
): Promise<string> {
  let fullResponse = '';
 
  const params: Anthropic.MessageStreamParams = {
    model: 'claude-sonnet-4-6',
    max_tokens: 2048,
    messages: [{ role: 'user', content: userMessage }],
  };
  if (systemPrompt) params.system = systemPrompt;
 
  const stream = anthropic.messages.stream(params);
 
  stream.on('text', (text) => {
    process.stdout.write(text);
    fullResponse += text;
  });
 
  await stream.finalMessage();
  console.log();
  return fullResponse;
}
 
streamChat(
  'Explain the TypeScript type system.',
  'Target an experienced developer audience.'
).then((result) => {
  console.log(`\nDone: ${result.length} characters`);
});

5. Tool Use: Converting Function Calling

This is the most involved part of the migration. The concepts are identical, but the schema format changes in one key place.

Schema Conversion

# OpenAI Function Calling
openai_tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a city.",
            "parameters": {           # ← "parameters" in OpenAI
                "type": "object",
                "properties": {
                    "location": {"type": "string"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["location"]
            }
        }
    }
]
 
# Claude Tool Use
claude_tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a city.",
        "input_schema": {            # ← "input_schema" in Claude
            "type": "object",
            "properties": {
                "location": {"type": "string"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    }
]

Tool Use Agent Loop

import anthropic
import json
from typing import Any
 
def process_tool_call(name: str, inputs: dict) -> Any:
    if name == "get_weather":
        return {"temperature": 22, "condition": "Sunny", "humidity": 60}
    return {"error": f"Unknown tool: {name}"}
 
def tool_use_agent(user_message: str) -> str:
    client = anthropic.Anthropic()
 
    tools = [
        {
            "name": "get_weather",
            "description": "Get current weather for a city.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "location": {"type": "string"},
                    "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
                },
                "required": ["location"]
            }
        }
    ]
 
    messages = [{"role": "user", "content": user_message}]
    MAX_TURNS = 8  # Always cap the number of tool round-trips
 
    for _ in range(MAX_TURNS):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            tools=tools,
            messages=messages
        )
 
        if response.stop_reason == "tool_use":
            # Add assistant response to history verbatim
            messages.append({"role": "assistant", "content": response.content})
 
            tool_results = []
            for block in response.content:
                if block.type != "tool_use":
                    continue
                try:
                    result = process_tool_call(block.name, block.input)
                    is_error = False
                except Exception as exc:
                    # Surface the failure instead of swallowing it
                    result = {"error": str(exc)}
                    is_error = True
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result),
                    "is_error": is_error,
                })
 
            messages.append({"role": "user", "content": tool_results})
            continue
 
        # Any other stop reason: always recover the text that was generated
        text = " ".join(b.text for b in response.content if b.type == "text")
        if response.stop_reason == "max_tokens":
            raise RuntimeError(
                f"Hit max_tokens mid-generation (recovered {len(text)} chars)"
            )
        return text
 
    raise RuntimeError(f"Tool loop exceeded {MAX_TURNS} round-trips")
 
print(tool_use_agent("What's the weather in Tokyo right now?"))

Pay attention to how the loop terminates. If you branch only on end_turn, a response that stops for max_tokens or stop_sequence falls through every branch. I wrote this as a bare while True with a fixed "Done" return early in my own migration, and truncated requests came back as empty successes — no exception, no log line, just a plausible-looking result. Capping the round-trips and handling every non-tool_use stop reason explicitly removed that whole class of silence.

Returning tool failures as is_error: true is deliberate too. Raising leaves the agent blind to what happened; swallowing lets it continue on a false premise. Handing the failure back gives Claude room to correct its arguments and try again.

6. Multimodal (Vision) Migration

import anthropic
import base64
from pathlib import Path
 
def analyze_image(image_path: str, question: str) -> str:
    client = anthropic.Anthropic()
 
    image_bytes = Path(image_path).read_bytes()
    b64_image = base64.standard_b64encode(image_bytes).decode("utf-8")
 
    ext_to_mime = {
        ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
        ".png": "image/png", ".gif": "image/gif", ".webp": "image/webp"
    }
    media_type = ext_to_mime.get(Path(image_path).suffix.lower(), "image/jpeg")
 
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": media_type,
                            "data": b64_image
                        }
                    },
                    {"type": "text", "text": question}
                ]
            }
        ]
    )
    return response.content[0].text
 
def analyze_image_url(image_url: str, question: str) -> str:
    """Analyze an image from a public URL."""
    client = anthropic.Anthropic()
 
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": [
                    {"type": "image", "source": {"type": "url", "url": image_url}},
                    {"type": "text", "text": question}
                ]
            }
        ]
    )
    return response.content[0].text

7. Production Error Handling and Retry Strategy

import anthropic
import time
import random
from typing import Optional
 
class ClaudeAPIClient:
    """Production-ready Claude API client with exponential backoff retries."""
 
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.client = anthropic.Anthropic()
        self.max_retries = max_retries
        self.base_delay = base_delay
 
    def chat(
        self,
        user_message: str,
        system_prompt: Optional[str] = None,
        model: str = "claude-sonnet-4-6",
        max_tokens: int = 2048
    ) -> str:
        for attempt in range(self.max_retries + 1):
            try:
                params = {
                    "model": model,
                    "max_tokens": max_tokens,
                    "messages": [{"role": "user", "content": user_message}]
                }
                if system_prompt:
                    params["system"] = system_prompt
 
                response = self.client.messages.create(**params)
                return response.content[0].text
 
            except anthropic.RateLimitError as e:
                if attempt < self.max_retries:
                    wait = self._backoff(attempt, e)
                    print(f"⚠️ Rate limit — waiting {wait:.1f}s (attempt {attempt + 1})")
                    time.sleep(wait)
                else:
                    raise
 
            except anthropic.APIStatusError as e:
                if e.status_code == 529 and attempt < self.max_retries:
                    # 529 = API overloaded — retryable
                    wait = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                    print(f"⚠️ Overloaded — waiting {wait:.1f}s")
                    time.sleep(wait)
                elif e.status_code in (400, 401, 403, 404):
                    raise  # Client errors — do not retry
                elif attempt < self.max_retries:
                    time.sleep(self.base_delay * (2 ** attempt))
                else:
                    raise
 
            except anthropic.APIConnectionError as e:
                if attempt < self.max_retries:
                    wait = self.base_delay * (2 ** attempt)
                    print(f"⚠️ Connection error — waiting {wait:.1f}s: {e}")
                    time.sleep(wait)
                else:
                    raise
 
        raise RuntimeError("Exceeded maximum retries")
 
    def _backoff(self, attempt: int, error) -> float:
        """Honour Retry-After when present, otherwise exponential backoff."""
        backoff = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
 
        response = getattr(error, "response", None)
        if response is None:
            return backoff
 
        retry_after = response.headers.get("retry-after")
        if retry_after is None:
            return backoff
 
        try:
            # Take whichever is longer: the server's instruction or our backoff
            return max(float(retry_after), backoff)
        except ValueError:
            # Retry-After can be an HTTP-date. If we can't parse it, fall back.
            return backoff
 
client = ClaudeAPIClient(max_retries=3)
result = client.chat(
    user_message="Describe the transformer architecture.",
    system_prompt="Explain clearly for an advanced developer audience."
)
print(result)

That _backoff method reflects a mistake I made and had to unwind. My first version read getattr(error, "retry_after", None), which reads naturally enough.

I installed the anthropic Python SDK (1.0.0) in a scratch VM, built a RateLimitError from a 429 response carrying retry-after: 7, and checked each access path:

Access path	Value returned
`hasattr(err, "retry_after")`	`False`
`getattr(err, "retry_after", None)`	`None`
`err.response.headers.get("retry-after")`	`"7"`

The exception's public attributes are body, message, request, request_id, response, status_code, type, and workspace_id — no retry_after anywhere. So the original code never read the value the server was explicitly sending, and quietly fell through to backoff on every single retry. No exception, nothing in the logs. Read rate-limit headers through response, not off the exception.

For more on this specific failure mode, see When Your Claude API Retry Logic Made Rate Limits Worse.

8. Bulk Migration Script

import ast
import re
from pathlib import Path
 
class OpenAIToClaudeMigrator:
    """Semi-automated converter for OpenAI API code."""
 
    # Match only the access expression, so trailing calls survive
    RESPONSE_ACCESS = re.compile(r'\.choices\[0\]\.message\.content')
 
    MODEL_MAP = {
        "gpt-4o": "claude-sonnet-4-6",
        "gpt-4o-mini": "claude-haiku-4-5-20251001",
        "gpt-4-turbo": "claude-sonnet-4-6",
        "gpt-4": "claude-sonnet-4-6",
        "gpt-3.5-turbo": "claude-haiku-4-5-20251001",
        "o1-preview": "claude-opus-4-6",
        "o1-mini": "claude-sonnet-4-6",
    }
 
    def migrate_file(self, source: str, dest: str) -> dict:
        code = Path(source).read_text(encoding="utf-8")
        result = self._transform(code)
        Path(dest).write_text(result["code"], encoding="utf-8")
        return result
 
    def _find_calls_without_max_tokens(self, code: str):
        """Return line numbers of messages.create( calls missing max_tokens.
 
        A whole-file substring check misses every call after the first one
        that happens to include max_tokens. Walking the AST is the only
        reliable way to inspect calls individually.
        """
        try:
            tree = ast.parse(code)
        except SyntaxError:
            return None
 
        missing = []
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            func = node.func
            if not (isinstance(func, ast.Attribute) and func.attr == "create"):
                continue
            owner = func.value
            if not (isinstance(owner, ast.Attribute) and owner.attr == "messages"):
                continue
            if not any(kw.arg == "max_tokens" for kw in node.keywords):
                missing.append(node.lineno)
        return sorted(missing)
 
    def _transform(self, code: str) -> dict:
        warnings = []
 
        # Imports
        code = re.sub(r'from openai import OpenAI', 'import anthropic', code)
        code = re.sub(r'import openai', 'import anthropic', code)
 
        # Client init
        code = re.sub(r'OpenAI\(api_key=([^)]+)\)', r'anthropic.Anthropic(api_key=\1)', code)
        code = re.sub(r'OpenAI\(\)', 'anthropic.Anthropic()', code)
 
        # Model names (quoted exact matches only)
        for old, new in self.MODEL_MAP.items():
            code = re.sub(rf"(['\"]){re.escape(old)}\1", rf"\g<1>{new}\g<1>", code)
 
        # Method call
        code = re.sub(r'\.chat\.completions\.create\(', '.messages.create(', code)
 
        # Response access (never inject anything into the converted code)
        hits = len(self.RESPONSE_ACCESS.findall(code))
        if hits:
            code = self.RESPONSE_ACCESS.sub('.content[0].text', code)
            warnings.append(f"Converted {hits} response access site(s): .choices[0].message.content → .content[0].text")
 
        # Missing max_tokens, per call site
        missing = self._find_calls_without_max_tokens(code)
        if missing is None:
            warnings.append("⚠️  Could not parse the converted code. Please review it by hand.")
        elif missing:
            lines = ", ".join(f"line {n}" for n in missing)
            warnings.append(f"⚠️  max_tokens missing on {len(missing)} call(s) ({lines}). Claude requires it.")
 
        # Function calling
        if '"functions"' in code or '"function_call"' in code:
            warnings.append("⚠️  Function Calling detected. Manual conversion to Tool Use is required (see Section 5).")
 
        return {"code": code, "warnings": warnings}
 
migrator = OpenAIToClaudeMigrator()
result = migrator.migrate_file("openai_app.py", "claude_app.py")
print("Conversion complete:")
for w in result["warnings"]:
    print(f"  {w}")

How this script used to break code silently

Two parts of the script above were rewritten after they bit me. Both failures are worth spelling out, because any bulk find-and-replace tool can reproduce them.

The first was response access. The original version annotated the conversion inline, which felt helpful at the time:

# The old version
code = code.replace(
    '.choices[0].message.content',
    '.content[0].text  # ← converted to Claude format'
)

Run that over an unremarkable piece of OpenAI code:

def summarize(text: str) -> str:
    resp = client.chat.completions.create(
        model="gpt-4o", max_tokens=512,
        messages=[{"role": "user", "content": text}],
    )
    return resp.choices[0].message.content.strip()

and you get:

    return resp.content[0].text  # ← converted to Claude format.strip()

The .strip() is now inside a comment. The nasty part is that this is not a syntax error — ast.parse() accepts it happily. Evaluating both expressions against the same response object gave:

Expression evaluated	Value returned
Before: `r.choices[0].message.content.strip()`	`'POSITIVE'`
After: `r.content[0].text # ← converted to Claude format.strip()`	`' POSITIVE '`

The whitespace now flows downstream. If that value feeds a label comparison or a json.loads(), you find out days later as a puzzling branch failure rather than as a migration bug. Keep annotations out of the emitted code and return them in warnings instead.

The second was the max_tokens check:

# The old version
if 'messages.create(' in code and 'max_tokens' not in code:
    warnings.append("⚠️  max_tokens is required in the Claude API.")

Give it a file where one of three calls specifies max_tokens and 'max_tokens' not in code evaluates to False, so the other two are never reported. Running both implementations against exactly that file:

Implementation	Missing calls detected
Substring check (old)	0 — no warning at all
AST walk (new)	2 — line 13 and line 20

Arguments to messages.create( are usually spread over several lines, so a line-by-line substring check misses them too. Walking ast.walk() for Call nodes turned out to be the shortest reliable route.

A bulk converter is the one tool whose output its author is most inclined to trust. When the migration spans dozens of files, budget time to run the result through ast.parse() and to read the diff with your own eyes.

9. Zero-Downtime Phased Rollout

Phase 1: Introduce an AI Gateway Layer

from enum import Enum
import anthropic
from openai import OpenAI
import os
 
class AIProvider(Enum):
    OPENAI = "openai"
    CLAUDE = "claude"
 
class AIGateway:
    """Transparent gateway that routes between OpenAI and Claude."""
 
    def __init__(self, provider: AIProvider = AIProvider.OPENAI, rollout: float = 0.0):
        self.primary = provider
        self.rollout = rollout  # Fraction of traffic routed to Claude (0.0–1.0)
        self.openai = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
        self.claude = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
 
    def chat(self, user_message: str, system: str = "", **kwargs) -> dict:
        import random
        use_claude = self.primary == AIProvider.CLAUDE or random.random() < self.rollout
 
        if use_claude:
            return self._claude(user_message, system, **kwargs)
        return self._openai(user_message, system, **kwargs)
 
    def _claude(self, msg, system, **kw):
        params = {
            "model": kw.get("model", "claude-sonnet-4-6"),
            "max_tokens": kw.get("max_tokens", 2048),
            "messages": [{"role": "user", "content": msg}]
        }
        if system:
            params["system"] = system
        r = self.claude.messages.create(**params)
        return {
            "text": r.content[0].text, "provider": "claude",
            "input_tokens": r.usage.input_tokens,
            "output_tokens": r.usage.output_tokens
        }
 
    def _openai(self, msg, system, **kw):
        messages = ([{"role": "system", "content": system}] if system else [])
        messages.append({"role": "user", "content": msg})
        r = self.openai.chat.completions.create(
            model=kw.get("model", "gpt-4o"),
            messages=messages,
            max_tokens=kw.get("max_tokens", 2048)
        )
        return {
            "text": r.choices[0].message.content, "provider": "openai",
            "input_tokens": r.usage.prompt_tokens,
            "output_tokens": r.usage.completion_tokens
        }
 
# Phase 1: 10% canary
gw_p1 = AIGateway(rollout=0.10)
 
# Phase 2: 50% A/B test
gw_p2 = AIGateway(rollout=0.50)
 
# Phase 3: full migration
gw_p3 = AIGateway(provider=AIProvider.CLAUDE, rollout=1.0)

Phase 2: Measure and Compare

import time, statistics
from dataclasses import dataclass, field
from typing import List
 
@dataclass
class Metric:
    provider: str
    latency_ms: float
    input_tokens: int
    output_tokens: int
    cost_usd: float
 
class MigrationAnalyzer:
    PRICING = {
        "claude-sonnet-4-6": {"input": 3.0 / 1_000_000, "output": 15.0 / 1_000_000},
        "gpt-4o":            {"input": 5.0 / 1_000_000, "output": 15.0 / 1_000_000},
    }
 
    def __init__(self):
        self.metrics: List[Metric] = []
 
    def record(self, provider, model, latency_ms, in_tokens, out_tokens) -> Metric:
        p = self.PRICING.get(model, {"input": 0, "output": 0})
        cost = in_tokens * p["input"] + out_tokens * p["output"]
        m = Metric(provider, latency_ms, in_tokens, out_tokens, cost)
        self.metrics.append(m)
        return m
 
    def report(self) -> dict:
        def summarize(ms):
            if not ms:
                return {}
            lats = [m.latency_ms for m in ms]
            return {
                "count": len(ms),
                "avg_latency_ms": statistics.mean(lats),
                "p95_latency_ms": sorted(lats)[int(len(lats) * 0.95)],
                "total_cost_usd": sum(m.cost_usd for m in ms),
                "avg_cost_usd": statistics.mean(m.cost_usd for m in ms)
            }
 
        oai = summarize([m for m in self.metrics if m.provider == "openai"])
        cld = summarize([m for m in self.metrics if m.provider == "claude"])
        reduction = 0.0
        if oai.get("avg_cost_usd") and cld.get("avg_cost_usd"):
            reduction = (oai["avg_cost_usd"] - cld["avg_cost_usd"]) / oai["avg_cost_usd"] * 100
        return {"openai": oai, "claude": cld, "cost_reduction_pct": reduction}

10. TypeScript / Next.js Complete Migration Pattern

// lib/ai-client.ts
import Anthropic from '@anthropic-ai/sdk';
 
interface ChatOptions {
  systemPrompt?: string;
  maxTokens?: number;
  model?: string;
}
 
interface ChatResponse {
  text: string;
  inputTokens: number;
  outputTokens: number;
}
 
export class ClaudeClient {
  private client: Anthropic;
  private defaultModel: string;
 
  constructor(apiKey?: string, defaultModel = 'claude-sonnet-4-6') {
    this.client = new Anthropic({ apiKey });
    this.defaultModel = defaultModel;
  }
 
  async chat(userMessage: string, options: ChatOptions = {}): Promise<ChatResponse> {
    const { systemPrompt, maxTokens = 2048, model = this.defaultModel } = options;
 
    const params: Anthropic.MessageCreateParamsNonStreaming = {
      model,
      max_tokens: maxTokens,
      messages: [{ role: 'user', content: userMessage }],
    };
    if (systemPrompt) params.system = systemPrompt;
 
    const response = await this.client.messages.create(params);
    const textBlock = response.content.find((b) => b.type === 'text');
    if (!textBlock || textBlock.type !== 'text') throw new Error('No text response');
 
    return {
      text: textBlock.text,
      inputTokens: response.usage.input_tokens,
      outputTokens: response.usage.output_tokens,
    };
  }
 
  // Next.js App Router streaming via Web Streams API
  async streamToResponse(
    userMessage: string,
    options: ChatOptions = {}
  ): Promise<ReadableStream> {
    const { systemPrompt, maxTokens = 2048, model = this.defaultModel } = options;
 
    const params: Anthropic.MessageStreamParams = {
      model,
      max_tokens: maxTokens,
      messages: [{ role: 'user', content: userMessage }],
    };
    if (systemPrompt) params.system = systemPrompt;
 
    const stream = this.client.messages.stream(params);
 
    return new ReadableStream({
      async start(controller) {
        stream.on('text', (text) => controller.enqueue(new TextEncoder().encode(text)));
        stream.on('error', (err) => controller.error(err));
        await stream.finalMessage();
        controller.close();
      },
    });
  }
}
 
// app/api/chat/route.ts (Next.js App Router)
export async function POST(request: Request) {
  const { message, systemPrompt } = await request.json();
  const claude = new ClaudeClient();
  const stream = await claude.streamToResponse(message, { systemPrompt });
  return new Response(stream, {
    headers: { 'Content-Type': 'text/plain; charset=utf-8' },
  });
}

Prompt design habits carry over intact after the migration. See Claude AI Prompt Engineering Techniques: 2026 for patterns that apply to both APIs.

What the migration taught me that the docs don't mention

The conversion patterns above cover the mechanics. Let me close with a few things I only learned after moving my own services from OpenAI to the Claude API in production.

I run Dolice Labs, a set of AI technical blogs with an automated content-formatting pipeline, as an indie developer. For a long time its backend leaned on OpenAI-based summarization and formatting logic, and the description generator for a small AdMob-monetized app shared the same lineage. When I moved them to the Claude API, the hard part wasn't the code conversion — it was the behavioral differences that surfaced quietly afterward.

`max_tokens` directly shapes output quality

OpenAI tolerates omitting max_tokens; Claude requires it. The body already notes this, but in practice the surprise was that a placeholder limit chosen during porting resurfaced later as truncated summaries. Right after the cutover, formatting tasks intermittently lost their tails, and the cause was a too-small limit I had left in during the port.

Here are the values that proved stable in my environment, by task:

Task	Feel before (OpenAI)	Recommended max_tokens (Claude)	Truncation
Classification / labeling	Fine unset	256	None
Article summary (~400 chars)	Fine unset	1,024	Sporadic below 512
Long-form formatting	~2,048	4,096	Tail loss at 2,048

Raising the ceiling does not inflate the bill when the actual output is short, so erring on the generous side is safe. That is my takeaway after the migration.

Isolating `system` alone improved instruction-following

On OpenAI I used to fold role instructions into the first message. Claude expects them in the system parameter. That looks like a formatting nicety, but moving them over changed how reliably instructions were followed. In my formatting pipeline, format drift — ignoring the heading hierarchy I had specified, for instance — visibly dropped. Once I consolidated tone and constraints into system and routed only the variable input through user, things stabilized without lowering the temperature.

Cost intuition is not captured by the per-token price table

It is tempting to judge the migration from the unit-price table, but the bill is driven by average tokens per request and how well caching applies. In my automation, pinning the shared instructions in system and combining that with prompt caching cut the billed input tokens by roughly 40-60% in practice. How you handle the repeated, fixed portion matters more to the monthly cost than the headline unit price. See the Claude API Cost Optimization Guide and Semantic Cache Production Design for the details.

A migration looks like a one-time excavation, but in hindsight it was a process of getting used to small behavioral differences while running in production. I hope it spares fellow developers a few of the stumbles.

Summary: Your Migration Checklist

Migrating from OpenAI to Claude is well worth the effort for most teams. Here's your final checklist:

Before writing code: Set ANTHROPIC_API_KEY, add max_tokens to every call, and move system prompts to the top-level system parameter.

During migration: Use the AIGateway pattern for staged rollout. Run the MigrationAnalyzer to measure latency and cost in parallel. Convert Function Calling to Tool Use carefully using the patterns in Section 5.

After going live: Monitor error logs for rate limit (429) and overload (529) events. Verify that the retry logic in Section 7 handles them gracefully. Keep the OpenAI key available as a rollback option for at least 30 days.

One last thing: once the migration settles, go back and read every line your conversion script touched. As Section 8 showed, the breakages that never raise a syntax error are the ones that surface weeks later.

For the next step, the Claude API Cost Optimization Guide and prompt caching are where a migration pays for itself fastest.

The 200K-token context window and strong long-document reasoning are available the moment you flip the switch — take advantage of them from day one. Thank you for reading this far; I hope a section or two saves you a stumble.

Thank You for Reading

Claude Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.

✦Copy-paste ready implementation code
✦New advanced guides published daily
✦$5/mo or $10 for lifetime access

View Membership →