How to Migrate From OpenAI to Claude Without Breaking Your Users (2026)

# OpenAI:
gpt-5-6:        $2.50 / 1M input     $10.00 / 1M output
gpt-5-6-mini:   $0.15 / 1M input      $0.60 / 1M output

# Anthropic Claude:
claude-opus-5:      $15.00 / 1M input    $75.00 / 1M output   (with 90% cache discount)
claude-sonnet-4-6:   $3.00 / 1M input    $15.00 / 1M output   (with 90% cache discount)
claude-haiku-4-5:    $0.25 / 1M input     $1.25 / 1M output   (with 90% cache discount)

# Before (OpenAI):
from openai import OpenAI
client = OpenAI()

response = client.chat.completions.create(
    model="gpt-5-6",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What's 2+2?"}
    ]
    # max_tokens optional
)
answer = response.choices[0].message.content

# After (Claude):
from anthropic import Anthropic
client = Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,                          # REQUIRED
    system="You are a helpful assistant.",    # Separate parameter
    messages=[
        {"role": "user", "content": "What's 2+2?"}
    ]
)
answer = response.content[0].text             # Content is a list of blocks

def extract_text(response):
    """Extract the assistant's text from a Claude response, handling mixed blocks."""
    text_blocks = [b.text for b in response.content if b.type == "text"]
    return "".join(text_blocks)

def extract_tool_calls(response):
    """Get any tool_use blocks from the response."""
    return [b for b in response.content if b.type == "tool_use"]

# Now works whether response has plain text, tool calls, or both:
text = extract_text(response)
tool_calls = extract_tool_calls(response)

def normalize_messages_for_claude(openai_messages):
    """Split OpenAI messages into Claude's (system, messages) shape."""
    system_prompt = ""
    claude_messages = []
    for m in openai_messages:
        if m["role"] == "system":
            # Multiple system messages — concatenate (Claude has one system slot)
            system_prompt = (system_prompt + "\n\n" + m["content"]).strip()
        else:
            claude_messages.append(m)
    return system_prompt, claude_messages

# OpenAI tool definition:
openai_tool = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"},
                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
            },
            "required": ["location"]
        }
    }
}

# Claude tool definition — flatter, no wrapper:
claude_tool = {
    "name": "get_weather",
    "description": "Get current weather for a location",
    "input_schema": {                    # Not "parameters"
        "type": "object",
        "properties": {
            "location": {"type": "string", "description": "City name"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
        },
        "required": ["location"]
    }
}

# Automated converter:
def openai_tool_to_claude(t):
    fn = t["function"]
    return {
        "name": fn["name"],
        "description": fn["description"],
        "input_schema": fn["parameters"],
    }

# OpenAI response with tool call:
{
    "choices": [{
        "message": {
            "role": "assistant",
            "content": None,
            "tool_calls": [{
                "id": "call_abc123",
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "arguments": '{"location": "SF", "unit": "celsius"}'  # STRING, not object
                }
            }]
        },
        "finish_reason": "tool_calls"
    }]
}

# Claude response with tool call — content blocks, not a separate array:
{
    "content": [
        {"type": "text", "text": "Let me check the weather for you."},
        {
            "type": "tool_use",
            "id": "toolu_abc123",
            "name": "get_weather",
            "input": {"location": "SF", "unit": "celsius"}   # ACTUAL OBJECT, not string
        }
    ],
    "stop_reason": "tool_use"
}

# OpenAI feeding tool result back:
messages.append({
    "role": "assistant",
    "tool_calls": [previous_tool_call]
})
messages.append({
    "role": "tool",
    "tool_call_id": "call_abc123",
    "content": '{"temperature": 22, "condition": "sunny"}'  # STRING
})

# Claude feeding tool result back — user message with tool_result block:
messages.append({
    "role": "assistant",
    "content": previous_response.content   # The whole content list from the assistant
})
messages.append({
    "role": "user",
    "content": [{
        "type": "tool_result",
        "tool_use_id": "toolu_abc123",
        "content": '{"temperature": 22, "condition": "sunny"}',   # Can be string or list
        "is_error": False                                            # Optional, defaults false
    }]
})

{
    "role": "user",
    "content": [{
        "type": "tool_result",
        "tool_use_id": "toolu_abc123",
        "content": "Weather API returned 503 — service temporarily unavailable. Try again in a few seconds.",
        "is_error": True   # Model recognizes this is a failure and often retries or explains
    }]
}

# OpenAI: strict JSON schema response format
response = openai_client.chat.completions.create(
    model="gpt-5-6",
    messages=messages,
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "customer_intent",
            "strict": True,
            "schema": {
                "type": "object",
                "properties": {
                    "intent": {"type": "string", "enum": ["refund", "cancel", "info"]},
                    "confidence": {"type": "number"},
                    "urgency": {"type": "string", "enum": ["low", "medium", "high"]}
                },
                "required": ["intent", "confidence", "urgency"],
                "additionalProperties": False
            }
        }
    }
)
result = json.loads(response.choices[0].message.content)


# Claude: force tool_use with the target schema
CLASSIFY_TOOL = {
    "name": "classify_intent",
    "description": "Extract the customer's intent and urgency from their message.",
    "input_schema": {
        "type": "object",
        "properties": {
            "intent": {"type": "string", "enum": ["refund", "cancel", "info"]},
            "confidence": {"type": "number"},
            "urgency": {"type": "string", "enum": ["low", "medium", "high"]}
        },
        "required": ["intent", "confidence", "urgency"]
    }
}

response = anthropic_client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=[CLASSIFY_TOOL],
    tool_choice={"type": "tool", "name": "classify_intent"},   # FORCE this specific tool
    messages=messages
)

# Extract the tool call's input — that's your structured output:
tool_use_block = next(b for b in response.content if b.type == "tool_use")
result = tool_use_block.input   # Already a parsed dict, not a JSON string

# OpenAI SDK streaming:
stream = openai_client.chat.completions.create(
    model="gpt-5-6",
    messages=messages,
    stream=True
)
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end="", flush=True)


# Anthropic SDK streaming — similar pattern via .stream() helper:
with anthropic_client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=messages
) as stream:
    for text in stream.text_stream:      # Convenience: iterates text deltas only
        print(text, end="", flush=True)

    final_message = stream.get_final_message()   # Full response after stream ends

# OpenAI SSE frame:
data: {"choices":[{"delta":{"content":"Hello"}}]}

data: {"choices":[{"delta":{"content":" world"}}]}

data: [DONE]


# Claude SSE frame — with event type:
event: message_start
data: {"type":"message_start","message":{...}}

event: content_block_start
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}

event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" world"}}

event: content_block_stop
data: {"type":"content_block_stop","index":0}

event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}

event: message_stop
data: {"type":"message_stop"}

# Claude tool call streaming — buffer input_json_delta chunks per content block:
async for event in stream:
    if event.type == "content_block_start" and event.content_block.type == "tool_use":
        tool_buffer = {
            "id": event.content_block.id,
            "name": event.content_block.name,
            "input_json": ""
        }
    elif event.type == "content_block_delta" and event.delta.type == "input_json_delta":
        tool_buffer["input_json"] += event.delta.partial_json
    elif event.type == "content_block_stop":
        # Now safe to parse — full JSON assembled
        tool_input = json.loads(tool_buffer["input_json"])
        await execute_tool(tool_buffer["name"], tool_input)

async def process_user_request(user_message, user_context):
    # Primary: OpenAI response, this is what the user sees
    primary_response = await openai_client.chat.completions.create(
        model="gpt-5-6",
        messages=user_context.messages + [{"role": "user", "content": user_message}]
    )

    # Shadow: fire Claude in the background, log the response, don't return it
    asyncio.create_task(shadow_evaluate(
        user_message=user_message,
        user_context=user_context,
        primary_response=primary_response,
    ))

    return primary_response

async def shadow_evaluate(user_message, user_context, primary_response):
    try:
        system_prompt, claude_messages = normalize_messages_for_claude(
            user_context.messages + [{"role": "user", "content": user_message}]
        )
        shadow_response = await anthropic_client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=system_prompt,
            messages=claude_messages,
        )

        # Log both responses for offline comparison
        db.insert("shadow_evals", {
            "user_message": user_message,
            "primary_provider": "openai",
            "primary_response": primary_response.choices[0].message.content,
            "primary_tokens": primary_response.usage.total_tokens,
            "primary_cost_usd": calc_openai_cost(primary_response.usage),
            "shadow_provider": "anthropic",
            "shadow_response": extract_text(shadow_response),
            "shadow_tokens": shadow_response.usage.input_tokens + shadow_response.usage.output_tokens,
            "shadow_cost_usd": calc_anthropic_cost(shadow_response.usage),
            "timestamp": datetime.utcnow(),
        })
    except Exception as e:
        # Never let shadow errors affect user experience
        log.warning(f"Shadow eval failed: {e}")

JUDGE_PROMPT = """Compare these two AI responses to the same user question.
Score which one better addresses the question. Return JSON only.

User question: {user_message}

Response A (from {provider_a}): {response_a}

Response B (from {provider_b}): {response_b}

{{
  "better": "A" | "B" | "tie",
  "reasoning": "",
  "accuracy_a": <1-5>,
  "accuracy_b": <1-5>,
  "helpfulness_a": <1-5>,
  "helpfulness_b": <1-5>
}}"""

# Randomize A/B assignment to avoid position bias
def compare_via_judge(pair):
    a, b = random.sample([pair.primary, pair.shadow], 2)
    judge_result = judge_client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=200,
        messages=[{"role": "user", "content": JUDGE_PROMPT.format(...)}]
    )
    # Parse and record which underlying provider won

ROLLOUT_CONFIG = {
    "customer_support_agent": {
        "claude_percentage": 5,       # Start at 5%
        "assignment_key": "user_id",  # Consistent per user
    },
    "content_generation": {
        "claude_percentage": 0,       # Not migrating yet
        "assignment_key": "user_id",
    },
}

def get_provider_for_request(feature, user_id):
    config = ROLLOUT_CONFIG[feature]
    # Consistent hash: same user always gets same provider (until % changes)
    bucket = hash(f"{feature}:{user_id}") % 100
    if bucket < config["claude_percentage"]:
        return "anthropic"
    return "openai"

# In your handler:
def handle_request(user_message, feature, user_id):
    provider = get_provider_for_request(feature, user_id)
    metrics.emit("migration.provider_assignment", 1, tags={
        "feature": feature, "provider": provider
    })
    if provider == "anthropic":
        return process_with_claude(user_message)
    else:
        return process_with_openai(user_message)

# Feature flag service reads config dynamically:
def get_current_rollout_percentage(feature):
    # Redis, LaunchDarkly, or similar — updated by ops without deploy
    return feature_flags.get_int(f"migration.{feature}.claude_percentage", default=0)

def instrumented_call(provider, model, feature, user_id, call_fn):
    start = time.time()
    tags = {"provider": provider, "model": model, "feature": feature}
    try:
        response = call_fn()
        metrics.emit("ai.request.success", 1, tags=tags)
        metrics.emit("ai.latency.total_ms", (time.time() - start) * 1000, tags=tags)
        tokens = get_total_tokens(response, provider)
        cost = calculate_cost(response, provider, model)
        metrics.emit("ai.tokens.total", tokens, tags=tags)
        metrics.emit("ai.cost.usd", cost, tags=tags)
        return response
    except Exception as e:
        metrics.emit("ai.request.error", 1, tags={**tags, "error_type": type(e).__name__})
        raise