Claude API Streaming in Python: Production Setup with OpenAI SDK

Last verified: July 21, 2026

Short path: Claude rates on Claude API. Non-streaming baseline: Claude Python example.

Claude API streaming delivers tokens incrementally over Server-Sent Events (SSE) instead of waiting for a full completion JSON. Copilot UIs, coding assistants, and chat products depend on streaming for perceived latency — users see the first word in hundreds of milliseconds even when total generation takes ten seconds.

You can stream Claude Sonnet and Claude Opus through LumeAPI with the OpenAI Python SDK by setting base_url to https://api.lumeapi.site/v1 and stream=True. The same client can call GPT or Gemini by changing only model.

LumeAPI is an independent gateway — not Anthropic. Test tool calling, long contexts, and error handling on your heaviest prompts before production cutover.

Quick Answer

1. Install openai (Python 3.10+ recommended).
2. Configure OpenAI(api_key=..., base_url="https://api.lumeapi.site/v1").
3. Call client.chat.completions.create(..., stream=True).
4. Iterate chunks: chunk.choices[0].delta.content.
5. Default streaming chat to claude-sonnet-4-6; reserve claude-opus-4-8 for escalation.
6. Set client timeout (60–120s), cap max_tokens, abort upstream when HTTP client disconnects.

Streaming does not add a per-token surcharge — you pay for input and output tokens generated.

Sonnet vs Opus for streaming cost and UX

Long streams on Opus multiply output cost. A 2,000-token stream on Sonnet catalog output ≈ $0.018; same on Opus ≈ $0.030 — acceptable for rare escalations, expensive as a default.

Depth on tier choice: Claude Sonnet vs Opus pricing.

Minimal streaming example

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["LUMEAPI_KEY"],
    base_url="https://api.lumeapi.site/v1",
    timeout=120.0,
)

def stream_claude(messages, model="claude-sonnet-4-6"):
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=1024,
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

if __name__ == "__main__":
    for token in stream_claude([{"role": "user", "content": "Explain SSE in one paragraph."}]):
        print(token, end="", flush=True)
    print()

Collect full text while streaming (logging / usage)

def stream_claude_collect(messages, model="claude-sonnet-4-6") -> str:
    parts: list[str] = []
    stream = client.chat.completions.create(
        model=model,
        messages=messages,
        max_tokens=1024,
        stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            parts.append(delta)
    return "".join(parts)

Some compatible endpoints attach usage only on the final chunk when stream_options are supported — if absent, estimate from your tokenizer or log approximate length for dashboards.

FastAPI SSE endpoint (production sketch)

import os
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()
client = OpenAI(
    api_key=os.environ["LUMEAPI_KEY"],
    base_url="https://api.lumeapi.site/v1",
    timeout=120.0,
)

class ChatRequest(BaseModel):
    messages: list
    model: str = "claude-sonnet-4-6"
    max_tokens: int = 1024

@app.post("/v1/claude/stream")
async def claude_stream(body: ChatRequest):
    if body.max_tokens > 4096:
        raise HTTPException(400, "max_tokens too high")

    def event_generator():
        try:
            stream = client.chat.completions.create(
                model=body.model,
                messages=body.messages,
                max_tokens=body.max_tokens,
                stream=True,
            )
            for chunk in stream:
                text = chunk.choices[0].delta.content
                if text:
                    # Escape newlines for naive SSE clients
                    yield f"data: {text}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as exc:
            yield f"event: error\ndata: {exc}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

Disable reverse-proxy buffering (X-Accel-Buffering: no on nginx) or tokens arrive in one batch.

Abort when the client disconnects

Wasted generation still bills output tokens. Tie stream lifetime to the HTTP connection:

import threading

def stream_with_cancel(messages, cancel_event: threading.Event):
    stream = client.chat.completions.create(
        model="claude-sonnet-4-6",
        messages=messages,
        stream=True,
    )
    for chunk in stream:
        if cancel_event.is_set():
            break
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

In FastAPI, watch request.is_disconnected() in a background task and set cancel_event.

Interrupted streams: recover interrupted LLM stream.

Tiered routing: Sonnet stream with Opus escalation

MODELS = {
    "default": "claude-sonnet-4-6",
    "hard": "claude-opus-4-8",
}

def stream_with_escalation(messages, complexity: str = "default"):
    model = MODELS["hard"] if complexity == "hard" else MODELS["default"]
    yield from stream_claude(messages, model=model)

Log complexity and model — reconcile with Usage. Do not escalate on habit; use eval gates or explicit user tier.

Tool calling + streaming

Tool use while streaming is provider-sensitive. Baseline non-streaming patterns: use Claude with OpenAI SDK.

Production advice:

• Test tool schemas in staging with real payloads
• Fall back to non-streaming for tool-heavy agent steps if chunks omit tool_calls reliably
• Keep GPT + Claude on one SDK for operational simplicity — multi-model Python

Rate limits and 429 during Claude streams

Streaming counts toward concurrency limits. On 429 before stream start, use bounded backoff — OpenAI 429 guide applies to compatible endpoints.

Cap parallel Claude streams per tenant during traffic spikes.

Production checklist

• [ ] API key only on server — never in browser bundles
• [ ] Default claude-sonnet-4-6 for streaming surfaces
• [ ] max_tokens cap per route (e.g. 1024 chat, 4096 long-form)
• [ ] Client timeout ≥ p99 generation time
• [ ] Cancel upstream on client disconnect
• [ ] Disable proxy buffering for SSE
• [ ] Log model, latency, approximate output length
• [ ] Load-test longest allowed prompt before launch

FAQ

Native Anthropic SDK vs OpenAI SDK through LumeAPI?

If you already standardized on OpenAI client for GPT, keep one integration for GPT + Claude + Gemini via LumeAPI. Native Anthropic SDK is appropriate for Anthropic-only direct integrations.

Does streaming change Claude pricing?

No — same per-million input/output rates. Longer answers cost more because of more output tokens, not because of SSE.

Sonnet or Opus for code streaming?

Start Sonnet. Escalate to Opus when evals show systematic failures on your repo’s patterns — Sonnet vs Opus.

How does this compare to GPT Node streaming?

Same SSE chunk shape — GPT Node.js streaming.

Interrupted streams?

See recover interrupted LLM stream. Prefer idempotent resume UX over blind full retry.

Can I stream cheap Claude alternatives?

Compare cheap Claude API rates vs Sonnet default — Opus is rarely the cost-optimized choice.

Sources and verification

• Claude API, cheap Claude API — July 21, 2026.
• OpenAI Python SDK streaming examples.
• Related: Claude Python example, GPT Python example.