Claude API Response Streaming in Python: A Guide

Streaming lets your Python app receive Claude's response as it's generated, token by token, instead of waiting for the full completion before showing anything. This matters for chat UIs, CLI tools, and any interface where perceived latency affects user experience — a response that starts appearing in 300ms feels far faster than one that arrives all at once after 8 seconds, even if the total generation time is identical.

The short answer: you set stream=True when calling the Messages API, then iterate over an event stream that emits typed events (content_block_delta, message_stop, etc.) as text arrives. Below is exactly how to do that correctly, plus the edge cases that trip people up — tool use during streaming, error handling mid-stream, and reconstructing the full message when you need it.

Streaming with the Anthropic Python SDK

If you're using anthropic, the official Python SDK, streaming is built in via a context manager:

import anthropic

client = anthropic.Anthropic(api_key="your-api-key")

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain event loops in Python"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final_message = stream.get_final_message()

stream.text_stream gives you just the text deltas, which is what you want for a simple print-as-you-go loop. stream.get_final_message() blocks until the stream finishes and returns the assembled Message object with full usage stats, stop reason, and content blocks — useful if you need to log token counts after the fact.

Handling raw stream events

Sometimes you need more than plain text — for example, tracking tool calls or content block boundaries. The SDK exposes the raw event stream too:

with client.messages.stream(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "List three testing frameworks"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            print(event.delta.text, end="", flush=True)
        elif event.type == "message_stop":
            print("\n--- done ---")

The main event types you'll encounter:

• message_start — the stream has begun, includes initial usage
• content_block_start — a new content block (text or tool_use) is starting
• content_block_delta — incremental text or partial JSON for tool inputs
• content_block_stop — a block is complete
• message_delta — top-level fields like stop_reason are updated
• message_stop — the stream is finished

Streaming without the SDK (raw SSE)

Claude's streaming API uses Server-Sent Events over HTTP. If you want to avoid the SDK dependency — or you're building against a proxy like SubToAPI that speaks the same Messages API shape — you can consume the stream directly with requests or httpx:

import httpx
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-3-5-sonnet-20241022",
    "max_tokens": 1024,
    "stream": True,
    "messages": [{"role": "user", "content": "Write a haiku about Python"}],
}

with httpx.stream("POST", url, headers=headers, json=payload, timeout=60) as response:
    for line in response.iter_lines():
        if line.startswith("data: "):
            data = line[len("data: "):]
            if data.strip() == "[DONE]":
                break
            event = json.loads(data)
            if event.get("type") == "content_block_delta":
                delta = event["delta"]
                if delta.get("type") == "text_delta":
                    print(delta["text"], end="", flush=True)

This is the pattern to know if you're calling the API from a framework where the SDK's async model doesn't fit cleanly, or if you're debugging why a proxy or gateway is mangling the stream.

Async streaming

For FastAPI, aiohttp-based services, or anything using asyncio, use AsyncAnthropic:

import asyncio
import anthropic

async def main():
    client = anthropic.AsyncAnthropic(api_key="your-api-key")
    async with client.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Summarize REST vs GraphQL"}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(main())

This is the version you want if you're piping the stream directly into a FastAPI StreamingResponse for a frontend — you yield each chunk as it arrives instead of accumulating it in memory.

Common streaming pitfalls

Not flushing stdout. If you print without flush=True (or sys.stdout.flush()), Python buffers output and you won't see incremental text in a terminal — it'll all appear at once, defeating the purpose.

Losing tool_use blocks in the delta stream. Tool calls stream their JSON arguments incrementally as input_json_delta events, not as one complete object. If you're building an agent loop, accumulate the partial JSON string per content block and parse it only once you get content_block_stop.

Not handling mid-stream errors. A stream can fail partway through (rate limits, overloaded errors) and emit an error event instead of completing normally. Wrap your stream loop in a try/except and check stream.get_final_message().stop_reason for "error" or missing completion.

Assuming streaming reduces cost. It doesn't — you pay for the same input and output tokens whether streamed or not. Streaming only changes when you receive the tokens, not how many.

Streaming through SubToAPI

If you're distributing Claude access inside a product — giving each customer or internal team their own API key instead of sharing one Anthropic credential — SubToAPI sits between your app and Claude and exposes the same streaming behavior through sub_live_... keys. The request shape matches the Messages API, so the Python code above works unchanged: set stream: true, point base_url at https://api.subtoapi.app/v1, and you get per-key usage metadata alongside the stream, useful for billing or rate-limiting individual users. See /docs/streaming for the exact endpoint and event format, or /docs/quickstart to get a key from the free trial.

questions

Does streaming reduce token costs? No. Token usage and pricing are identical whether you stream or wait for the full response — streaming only changes delivery timing, not billing.

Can I stream tool use (function calling) responses? Yes. Tool inputs arrive as input_json_delta events across multiple chunks. Accumulate the partial JSON string per content block and parse it once you receive content_block_stop.

What's the difference between text_stream and iterating raw events? text_stream gives you only the plain text deltas, ideal for simple print-as-you-go output. Iterating raw events (content_block_delta, message_stop, etc.) gives you full control, needed for tracking tool calls or usage metadata mid-stream.