How to Stream Claude Responses in Python

Streaming Claude Responses in Python

If you're building anything beyond a simple script — a chatbot, a CLI tool, a backend that feeds a frontend UI — you don't want to wait for Claude to generate a full response before showing anything to the user. Streaming solves that by sending the response token-by-token (technically, chunk-by-chunk) as it's generated, so your app can display text as it arrives instead of staring at a spinner for 10+ seconds.

In Python, there are three practical ways to stream Claude responses: using the official anthropic SDK's streaming helper, consuming raw server-sent events (SSE) yourself, or going through an async client for non-blocking I/O in a web server. This guide covers all three, with working code you can copy directly.

Option 1: The Anthropic SDK's Streaming Helper (Easiest)

The official Python SDK wraps streaming in a context manager that handles the SSE parsing for you. This is the fastest path if you just want text on screen.

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 streaming in one paragraph."}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final_message = stream.get_final_message()

Key points:

• stream.text_stream yields plain text chunks — no manual JSON parsing needed.
• stream.get_final_message() gives you the complete Message object after streaming finishes, including token usage and stop reason.
• The end="", flush=True combo in print() is what makes the output appear incrementally in your terminal instead of all at once.

This pattern works the same way whether you're hitting Anthropic's API directly or a compatible gateway endpoint — the SDK just needs a base URL and key.

Option 2: Raw SSE with requests or httpx

Sometimes you don't want the SDK dependency, or you're integrating with a service that exposes a streaming endpoint without a Python client. In that case you parse server-sent events manually.

import httpx
import json

url = "https://api.subtoapi.app/v1/messages"
headers = {
    "Authorization": "Bearer sub_live_xxx",
    "Content-Type": "application/json",
}
payload = {
    "model": "claude-3-5-sonnet-20241022",
    "max_tokens": 1024,
    "stream": True,
    "messages": [{"role": "user", "content": "Write a haiku about async code."}],
}

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

This is the same SSE format Anthropic's API uses, so if you're already pointed at /v1/messages on SubToAPI (which proxies your Claude access behind an API key) the parsing logic is identical. See /docs/streaming and /docs/messages for the full event schema — events include message_start, content_block_delta, message_delta, and message_stop, each with its own payload shape.

A few things that trip people up with manual SSE parsing:

• Lines can be empty (used as event separators) — skip them.
• Not every data: line is a text delta. If you're also using tool use (see /docs/tools), you'll get input_json_delta events for tool arguments, which need different handling than text deltas.
• Set a reasonable timeout — long generations with large max_tokens can run well past the default timeout in most HTTP clients.

Option 3: Async Streaming for Web Servers

If you're streaming Claude's response through to a frontend (e.g., via FastAPI and Server-Sent Events or WebSockets), you want an async client so one slow Claude call doesn't block your whole event loop.

import anthropic
import asyncio

async def stream_response(prompt: str):
    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": prompt}],
    ) as stream:
        async for text in stream.text_stream:
            yield text

async def main():
    async for chunk in stream_response("Summarize the benefits of async I/O."):
        print(chunk, end="", flush=True)

asyncio.run(main())

Wrap this generator in a FastAPI StreamingResponse and you have a non-blocking streaming endpoint your frontend can consume directly:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(prompt: str):
    return StreamingResponse(stream_response(prompt), media_type="text/plain")

Using a Hosted API Instead of Managing Keys Directly

If you're building this inside a product — rather than a personal script — one issue shows up quickly: Anthropic's console doesn't give you per-application API keys, usage dashboards per key, or team seat management out of the box. SubToAPI sits in front of your existing Claude access and exposes it as a standard HTTPS API with sub_live_... keys, so each service or teammate gets its own scoped key, and streaming, tool use, and usage metadata all work the same way shown above — same /v1/messages endpoint, same SSE event format.

Setup is a signup plus key creation: see /docs/quickstart for the five-minute version. Plans start at €9/month for solo use, with team and scale tiers at €19 and €49 per seat (see /pricing), and there's a free trial if you want to test streaming against your own code before committing.

questions

Does streaming cost more than a regular Claude API call? No. You're billed on the same input/output token usage either way — streaming only changes how the response is delivered (incrementally vs. all at once), not what you're charged for.

Why is my stream hanging or timing out in Python? Usually it's an HTTP client timeout that's too short for long generations, or you're not flushing stdout (print(..., flush=True)). Also check you're setting "stream": True in the request payload if you're not using the SDK's dedicated streaming method.

Can I stream tool use (function calling) the same way as text? Mostly yes — tool use streams as input_json_delta events containing partial JSON for the tool's arguments, which you accumulate and parse once complete, rather than printing directly like text deltas. See /docs/tools for the event structure.