Claude API Streaming Response Implementation in Python

Streaming lets your Python application display Claude's response token-by-token instead of waiting for the entire completion to finish. This matters for any chat interface, CLI tool, or long-form generation feature where users expect to see text appear immediately rather than staring at a spinner for 10–30 seconds.

The short answer: Claude's API streams responses using server-sent events (SSE) over HTTP. In Python, you either use Anthropic's official SDK with stream=True, or you parse the raw SSE stream yourself with requests and iter_lines(). Both approaches give you a sequence of events you accumulate into the final text. Below is a working implementation of each, plus the event types you need to handle correctly.

Streaming with the Anthropic Python SDK

The official SDK abstracts away the SSE parsing and gives you a context manager that yields text deltas directly:

import anthropic

client = anthropic.Anthropic(api_key="sk-ant-...")

with client.messages.stream(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain TCP slow start"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

    final_message = stream.get_final_message()
    print("\n\nTokens used:", final_message.usage.output_tokens)

stream.text_stream only yields text deltas, which is enough for most UI use cases. If you need full control over events (tool calls, stop reasons, usage), iterate over stream directly instead of text_stream.

Handling raw SSE events manually

If you're not using the SDK — maybe you're proxying requests, logging raw events, or building your own client — you need to understand the event sequence Claude emits:

• message_start — the message object is created, contains initial usage info
• content_block_start — a new content block (text or tool_use) begins
• content_block_delta — incremental text or JSON fragments
• content_block_stop — the block is complete
• message_delta — stop reason and final usage update
• message_stop — stream finished

Here's a minimal parser using requests:

import json
import requests

url = "https://api.anthropic.com/v1/messages"
headers = {
    "x-api-key": "sk-ant-...",
    "anthropic-version": "2023-06-01",
    "content-type": "application/json",
}
payload = {
    "model": "claude-sonnet-4-5",
    "max_tokens": 1024,
    "stream": True,
    "messages": [{"role": "user", "content": "Write a haiku about rate limits"}],
}

with requests.post(url, headers=headers, json=payload, stream=True) as resp:
    for line in resp.iter_lines():
        if not line or not line.startswith(b"data: "):
            continue
        data = json.loads(line[6:])
        if data.get("type") == "content_block_delta":
            delta = data["delta"]
            if delta.get("type") == "text_delta":
                print(delta["text"], end="", flush=True)

This is the pattern to replicate for any language or framework that doesn't have an official SDK binding — Claude's streaming format is a predictable sequence of typed JSON events over SSE, so the parsing logic above translates directly to Node, Go, or Ruby with minimal changes.

Common implementation mistakes

Not buffering partial SSE lines. iter_lines() can split a single event across chunks on slow connections. If you see json.JSONDecodeError, buffer bytes until you hit a full data: ...\n\n block rather than assuming one line equals one event.

Ignoring message_delta for final usage. Input/output token counts arrive in two places: an estimate at message_start and the authoritative count at message_delta. If you're billing users per token, only trust the final message_delta payload.

Blocking the event loop in async contexts. If you're using httpx with asyncio, make sure you're using client.stream() as an async context manager and async for over lines — mixing sync requests into an async FastAPI handler will stall concurrent requests.

Not handling tool_use deltas separately. If your prompt includes tools, content_block_delta events for tool calls arrive as input_json_delta with partial JSON strings you need to concatenate and parse once the block closes — treating them like text deltas will corrupt the arguments.

A simpler path: streaming through SubToAPI

If you're building a product on top of Claude and want streaming without managing Anthropic SDK versions, SSE parsing edge cases, or separate billing per provider account, SubToAPI exposes your Claude access as a standard HTTPS API with the same streaming semantics, under your own sub_live_... key.

curl https://api.subtoapi.app/v1/messages \
  -H "Authorization: Bearer $SUBTOAPI_KEY" \
  -H "content-type: application/json" \
  -d '{
    "model": "claude-sonnet-4-5",
    "max_tokens": 1024,
    "stream": true,
    "messages": [{"role": "user", "content": "Stream me a short story"}]
  }'

Or from JavaScript, for teams building a Node backend alongside a Python one:

const res = await fetch("https://api.subtoapi.app/v1/messages", {
  method: "POST",
  headers: {
    Authorization: `Bearer ${process.env.SUBTOAPI_KEY}`,
    "content-type": "application/json",
  },
  body: JSON.stringify({
    model: "claude-sonnet-4-5",
    max_tokens: 1024,
    stream: true,
    messages: [{ role: "user", content: "Stream me a short story" }],
  }),
});

for await (const chunk of res.body) {
  process.stdout.write(Buffer.from(chunk).toString());
}

The event format matches what's described above, so the Python parsing code you already wrote works unchanged — you just point it at a different base URL and key. This is useful if multiple developers on your team each need their own API key and usage tracking without sharing one Anthropic account; see /pricing for Solo, Team, and Scale plans, or jump into the quickstart and streaming docs for full request/response examples.

Questions

Does streaming cost more than non-streaming requests? No. Token pricing is identical whether you stream or wait for the full response — streaming only changes delivery, not billing.

Can I cancel a stream mid-response to save tokens? Yes, closing the HTTP connection stops generation server-side. Output tokens generated up to that point are still billed, but you avoid paying for the rest of a response you don't need.

Why am I getting empty text_delta events with the SDK? This usually means you're iterating over the raw stream object instead of stream.text_stream, and some deltas belong to non-text content blocks like tool calls — check content_block_start for the block type before assuming it's text.