Claude API Streaming Implementation with Python

If you're searching for a Claude API streaming implementation with Python, you want token-by-token output in your terminal or app instead of waiting for the full response to generate. This matters for chat UIs, CLIs, and any interface where perceived latency affects user experience. Below is a complete working implementation using Python's requests library, plus the common mistakes that break streaming in production.

The short version: Claude's API supports streaming via Server-Sent Events (SSE) when you set "stream": true in your request body. You then read the response as a line-by-line event stream, parse each content_block_delta event, and print or forward the text as it arrives. The rest of this article covers the full implementation, error handling, and how to adapt it if you're calling Claude through an API wrapper like SubToAPI instead of directly.

Setting Up the Request

Streaming works over a standard HTTPS POST request. The difference is that instead of waiting for a single JSON response, the server keeps the connection open and sends a sequence of events as the model generates tokens.

Here's the baseline request structure with requests:

import requests
import json

url = "https://api.anthropic.com/v1/messages"
headers = {
    "x-api-key": "YOUR_API_KEY",
    "anthropic-version": "2023-06-01",
    "content-type": "application/json"
}
payload = {
    "model": "claude-sonnet-4-5",
    "max_tokens": 1024,
    "stream": True,
    "messages": [
        {"role": "user", "content": "Explain SSE streaming in two sentences."}
    ]
}

response = requests.post(url, headers=headers, json=payload, stream=True)

The critical part is stream=True on the requests.post() call itself — this tells the library not to buffer the entire response into memory before returning. If you forget this, you'll get the full response at once with no incremental output, even if your JSON payload says "stream": true.

Parsing the Event Stream

Claude's SSE stream emits several event types: message_start, content_block_start, content_block_delta, content_block_stop, message_delta, and message_stop. For most use cases, you only care about content_block_delta, which carries the actual text chunks.

def stream_claude_response(response):
    full_text = ""
    for line in response.iter_lines():
        if not line:
            continue
        decoded = line.decode("utf-8")
        if decoded.startswith("data: "):
            data_str = decoded[len("data: "):]
            if data_str.strip() == "[DONE]":
                break
            try:
                event = json.loads(data_str)
            except json.JSONDecodeError:
                continue

            if event.get("type") == "content_block_delta":
                delta = event.get("delta", {})
                text = delta.get("text", "")
                full_text += text
                print(text, end="", flush=True)

    print()
    return full_text

result = stream_claude_response(response)

Three things matter here:

• Use iter_lines(), not response.text. Accessing .text or .json() forces requests to read the whole body first, defeating the purpose of streaming.
• Skip empty lines. SSE uses blank lines as event separators; your parser needs to ignore them rather than throw errors.
• Flush stdout explicitly. print(text, end="", flush=True) is required in Python because stdout is buffered by default — without flush=True you'll still see output arrive in chunks instead of character by character.

Handling Errors Mid-Stream

A stream can fail partway through — rate limits, network drops, or an error event embedded in the SSE payload itself. Wrap the loop so a partial failure doesn't crash silently or leave you with truncated output you mistake for a complete answer:

def stream_claude_response(response):
    full_text = ""
    try:
        for line in response.iter_lines():
            if not line:
                continue
            decoded = line.decode("utf-8")
            if not decoded.startswith("data: "):
                continue
            data_str = decoded[len("data: "):]
            event = json.loads(data_str)

            if event.get("type") == "error":
                raise RuntimeError(event.get("error", {}).get("message", "stream error"))

            if event.get("type") == "content_block_delta":
                text = event.get("delta", {}).get("text", "")
                full_text += text
                print(text, end="", flush=True)

            if event.get("type") == "message_stop":
                break
    except requests.exceptions.ChunkedEncodingError:
        print("\n[stream interrupted — connection dropped]")
    finally:
        response.close()

    return full_text

Closing the response in finally avoids leaking open sockets when you're streaming many requests in a loop, which is easy to miss in scripts that run for hours.

Async Streaming for Web Backends

If you're exposing this in a FastAPI or similar backend, you want an async generator rather than blocking the event loop with synchronous requests calls. Using httpx instead:

import httpx

async def stream_claude(payload: dict, headers: dict):
    async with httpx.AsyncClient() as client:
        async with client.stream("POST", "https://api.anthropic.com/v1/messages",
                                  headers=headers, json=payload, timeout=60) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    data_str = line[len("data: "):]
                    if data_str.strip() == "[DONE]":
                        return
                    yield data_str

You can then forward each data_str directly to your own SSE endpoint using FastAPI's StreamingResponse, giving your frontend the same token-by-token experience without the client ever touching Anthropic's API directly.

Streaming Through an API Gateway

If your app already calls Claude through an API gateway rather than Anthropic's endpoint directly — for things like per-key usage tracking, rate limiting, or team billing — the parsing logic above stays identical. Only the URL and headers change. SubToAPI, for example, exposes the same Messages-compatible streaming format at https://api.subtoapi.app/v1/messages, so the iter_lines() loop and delta parsing you just wrote works unmodified — you just swap in your sub_live_... key and point the client at the new host. See /docs/streaming for request/response details and /docs/quickstart if you're setting this up for the first time.

The main advantage of routing through a gateway layer is that you get usage metadata per request (token counts, cost, which API key made the call) without instrumenting that logic yourself in every service that streams responses.

Common Pitfalls

• Buffering in requests without stream=True — the single most common mistake; the request "works" but arrives all at once.
• Printing with default print() flush behavior — looks broken even when the stream is working correctly.
• Not handling [DONE] or message_stop — leaves connections hanging.
• Reconstructing JSON from partial deltas instead of accumulating text — deltas are plain text fragments, not partial JSON objects, so don't try to parse each chunk as standalone JSON content.

Questions

Does Claude's streaming API use WebSockets? No. It uses Server-Sent Events (SSE) over a standard HTTP connection, which is simpler to implement and works with standard HTTP clients and proxies.

Can I stream Claude responses from a serverless function? Yes, as long as the platform supports long-lived HTTP responses (most do via streaming response objects), but check your platform's timeout limits for long generations.

Why does my stream print nothing until the end? This almost always means stream=True is missing from your requests.post() call, or you're calling .text/.json() on the response before iterating over lines.