Claude Streaming API in Python: Streaming Responses for Real-Time UX

Why Streaming Matters for Real-World Products

When a user submits a prompt and your app calls client.messages.create() without streaming, the HTTP connection stays open for the full generation time, and nothing arrives until Claude has finished writing every token. For a 400-token answer at typical generation speed, that is 3 to 8 seconds of silence. Users click away, assume the request failed, or double-submit.

Streaming solves this at the perception layer. With the claude streaming api, the first token arrives in under 200 ms (often 80 to 130 ms from Anthropic’s API edge). The rest of the text flows in continuously. Users read at roughly the same rate Claude writes, so the experience feels like watching a thoughtful person type, not waiting for a black box to respond.

This is not just a UI nicety. Engineering teams at chat products, coding assistants, and long-form generation tools consistently report that switching from blocking to streaming drops support tickets about “the app is broken” by 40 to 70 percent. Perceived reliability improves even when the median total latency is identical.

Who This Pattern Is For

• Teams building chat interfaces on top of Claude (customer service bots, copilots, internal tools).
• CLI tooling authors who want incremental output instead of a spinner while waiting.
• Backend engineers exposing Claude over HTTP to a React or plain JS frontend.
• Anyone who has already built a blocking Claude integration and wants to improve UX without a full rewrite.

If you are new to the Anthropic SDK, read Part 1: What Claude Can Do in Production first, then come back. This article assumes you can already make a basic blocking call.

How the Claude Streaming API Works Internally

Claude’s streaming API uses HTTP chunked transfer encoding under the hood. The server pushes a sequence of Server-Sent Events on the same HTTP/2 connection, each containing one event type and a JSON payload. The Anthropic Python SDK wraps all of that in a clean context manager so you never parse raw SSE frames yourself.

The Key Event Types

When you iterate over stream.text_stream from the SDK, the SDK has already filtered out the plumbing events and hands you pure text strings. But it helps to know what the raw stream contains:

The SDK Streaming Interface

The SDK exposes three ways to consume a stream:

1. stream.text_stream: an async/sync generator yielding raw text strings. Best for most use cases.
2. stream itself as an iterator: yields MessageStreamEvent objects if you need the raw event types above.
3. stream.get_final_message(): blocks until completion and returns a full Message object including usage. Useful when you need total token counts after streaming is done.

Setting Up the Project

Installation

The full requirements file is shown in the POC below. uvicorn is the ASGI server for FastAPI. python-dotenv loads your ANTHROPIC_API_KEY from a .env file during development.

Project Layout

streaming-poc/
  .env
  requirements.txt
  cli_stream.py          # Part 1: command-line streaming demo
  web_stream.py          # Part 2: FastAPI SSE endpoint
  static/
    index.html           # Part 3: browser client

Environment Setup

The Anthropic SDK reads ANTHROPIC_API_KEY from the environment automatically when you instantiate Anthropic(). Never hardcode the key in source files.

# .env
ANTHROPIC_API_KEY=sk-ant-...        # your key from console.anthropic.com

The Claude Streaming API: Full POC Code

This section contains three complete, runnable files. Copy them as-is into the project layout described above, set your .env, and run.

requirements.txt

anthropic>=0.28.0
fastapi>=0.111.0
uvicorn[standard]>=0.30.0
python-dotenv>=1.0.0

Part 1: CLI Streaming (cli_stream.py)

The CLI tool reads a prompt from the command line, opens a stream, and prints tokens as they arrive. It also prints usage stats when the stream ends.

"""
cli_stream.py
Usage: python cli_stream.py "Explain how HTTP chunked transfer encoding works."
"""

import sys
import os
import anthropic
from dotenv import load_dotenv

load_dotenv()


def stream_to_stdout(prompt: str, model: str = "claude-sonnet-4-6", max_tokens: int = 1024) -> None:
    """Stream a Claude response to stdout, flushing after each token."""
    client = anthropic.Anthropic()

    print(f"\n[Prompt]: {prompt}\n")
    print("[Response]:\n")

    try:
        with client.messages.stream(
            model=model,
            max_tokens=max_tokens,
            system="You are a clear, concise technical explainer. Use plain language.",
            messages=[{"role": "user", "content": prompt}],
        ) as stream:
            for text in stream.text_stream:
                print(text, end="", flush=True)

            # get_final_message() is safe to call after the loop; it collects
            # the terminal message_delta event that carries output token count.
            final = stream.get_final_message()

        print("\n")
        print(
            f"[Usage] input={final.usage.input_tokens} tokens  "
            f"output={final.usage.output_tokens} tokens  "
            f"stop_reason={final.stop_reason}"
        )

    except anthropic.APIStatusError as exc:
        print(f"\n[API Error {exc.status_code}]: {exc.message}", file=sys.stderr)
        sys.exit(1)
    except anthropic.APIConnectionError as exc:
        print(f"\n[Connection Error]: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python cli_stream.py \"<your prompt here>\"", file=sys.stderr)
        sys.exit(1)

    prompt_text = " ".join(sys.argv[1:])
    stream_to_stdout(prompt_text)

Sample CLI Run

$ python cli_stream.py "Explain how HTTP chunked transfer encoding works in 3 bullet points."

[Prompt]: Explain how HTTP chunked transfer encoding works in 3 bullet points.

[Response]:

- Chunked transfer encoding lets a server send an HTTP response body in a series
  of pieces ("chunks") without knowing the total size in advance. Each chunk is
  prefixed with its byte length in hexadecimal, followed by the data, then CRLF.

- The client reassembles chunks in order as they arrive. A zero-length chunk
  signals the end of the body, after which optional trailers and a final CRLF
  close the message.

- This is especially useful for streaming AI responses, database query results,
  and file downloads where the server begins writing before it has finished
  computing, reducing time-to-first-byte for the end user.

[Usage] input=42 tokens  output=117 tokens  stop_reason=end_turn

Part 2: FastAPI SSE Endpoint (web_stream.py)

Server-Sent Events (SSE) are plain HTTP. The server keeps the connection open and writes data: ...\n\n lines. The browser’s built-in EventSource API receives them. No WebSocket handshake, no additional protocol, no extra library on the frontend.

The generator function token_generator() is an async Python generator that yields SSE-formatted strings. FastAPI’s StreamingResponse accepts any async iterable and handles the chunked encoding automatically.

"""
web_stream.py
Start: uvicorn web_stream:app --reload --port 8000
Then open http://localhost:8000 in a browser (serves static/index.html).
SSE endpoint: GET /stream?prompt=your+text
"""

import asyncio
import json
import os
from typing import AsyncGenerator

import anthropic
from dotenv import load_dotenv
from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles

load_dotenv()

app = FastAPI(title="Claude Streaming Demo")

# Mount static files so /  serves static/index.html
app.mount("/static", StaticFiles(directory="static"), name="static")

_client = anthropic.Anthropic()


async def token_generator(prompt: str, model: str, max_tokens: int) -> AsyncGenerator[str, None]:
    """
    Async generator that opens a Claude stream and yields SSE-formatted strings.
    Each chunk is:   data: {"token": "..."}\n\n
    Final chunk is:  data: {"done": true, "input_tokens": N, "output_tokens": M}\n\n
    Error chunk is:  data: {"error": "..."}\n\n
    """
    try:
        # The anthropic SDK's stream() is synchronous under the hood.
        # We run it in a thread pool so we do not block the FastAPI event loop.
        loop = asyncio.get_running_loop()

        # Collect the full response in a separate thread using run_in_executor.
        # For true async streaming we use a queue to bridge the sync iterator
        # into the async world.
        queue: asyncio.Queue = asyncio.Queue()

        def _run_stream() -> None:
            """Runs in a thread; pushes tokens into the asyncio queue."""
            try:
                with _client.messages.stream(
                    model=model,
                    max_tokens=max_tokens,
                    system="You are a helpful technical assistant.",
                    messages=[{"role": "user", "content": prompt}],
                ) as stream:
                    for text in stream.text_stream:
                        # thread-safe put: schedule on the event loop
                        loop.call_soon_threadsafe(queue.put_nowait, {"token": text})

                    final = stream.get_final_message()
                    loop.call_soon_threadsafe(
                        queue.put_nowait,
                        {
                            "done": True,
                            "input_tokens": final.usage.input_tokens,
                            "output_tokens": final.usage.output_tokens,
                            "stop_reason": final.stop_reason,
                        },
                    )
            except anthropic.APIError as exc:
                loop.call_soon_threadsafe(queue.put_nowait, {"error": str(exc)})

        # Start the blocking stream in a background thread
        loop.run_in_executor(None, _run_stream)

        # Yield tokens as they appear in the queue
        while True:
            item = await queue.get()
            yield f"data: {json.dumps(item)}\n\n"
            if item.get("done") or item.get("error"):
                break

    except Exception as exc:
        yield f"data: {json.dumps({'error': str(exc)})}\n\n"


@app.get("/stream")
async def stream_endpoint(
    prompt: str = Query(..., description="The prompt to send to Claude"),
    model: str = Query(default="claude-sonnet-4-6"),
    max_tokens: int = Query(default=1024, ge=1, le=4096),
) -> StreamingResponse:
    """
    Server-Sent Events endpoint.  Accepts GET with ?prompt=...
    Returns text/event-stream that a browser EventSource can consume.
    """
    return StreamingResponse(
        token_generator(prompt, model, max_tokens),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",   # disables nginx buffering if behind a proxy
        },
    )


@app.get("/", response_class=HTMLResponse)
async def root() -> HTMLResponse:
    """Serve the demo UI from static/index.html."""
    with open("static/index.html", encoding="utf-8") as f:
        return HTMLResponse(f.read())

Part 3: Browser Client (static/index.html)

A minimal HTML page that uses the browser’s EventSource API. No JavaScript framework required. Tokens are appended to a <pre> element as they arrive.

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>Claude Streaming Demo</title>
  <style>
    body { font-family: system-ui, sans-serif; max-width: 800px; margin: 2rem auto; padding: 0 1rem; }
    textarea { width: 100%; height: 80px; font-size: 1rem; padding: 0.5rem; }
    button { margin-top: 0.5rem; padding: 0.5rem 1.5rem; font-size: 1rem; cursor: pointer;
             background: #0D5C73; color: white; border: none; border-radius: 4px; }
    button:disabled { opacity: 0.5; cursor: not-allowed; }
    #output { margin-top: 1.5rem; white-space: pre-wrap; background: #f5f5f5;
              padding: 1rem; border-radius: 6px; min-height: 120px; font-size: 0.95rem; }
    #stats { margin-top: 0.5rem; color: #555; font-size: 0.85rem; }
  </style>
</head>
<body>
  <h1>Claude Streaming Demo</h1>
  <textarea id="prompt" placeholder="Enter your prompt..."></textarea>
  <br>
  <button id="sendBtn" onclick="startStream()">Send</button>
  <div id="output"></div>
  <div id="stats"></div>

  <script>
    let activeSource = null;

    function startStream() {
      const prompt = document.getElementById("prompt").value.trim();
      if (!prompt) return;

      // Close any previous stream
      if (activeSource) { activeSource.close(); }

      const output = document.getElementById("output");
      const stats  = document.getElementById("stats");
      const btn    = document.getElementById("sendBtn");

      output.textContent = "";
      stats.textContent  = "";
      btn.disabled = true;

      // EventSource only supports GET; encode prompt in the query string
      const url = "/stream?prompt=" + encodeURIComponent(prompt);
      activeSource = new EventSource(url);

      activeSource.onmessage = function(event) {
        const data = JSON.parse(event.data);

        if (data.error) {
          output.textContent += "\n[Error]: " + data.error;
          activeSource.close();
          btn.disabled = false;
          return;
        }

        if (data.done) {
          stats.textContent =
            "Tokens used: " + data.input_tokens + " in / " + data.output_tokens + " out" +
            "  |  Stop reason: " + data.stop_reason;
          activeSource.close();
          btn.disabled = false;
          return;
        }

        if (data.token !== undefined) {
          output.textContent += data.token;
        }
      };

      activeSource.onerror = function() {
        output.textContent += "\n[Stream closed or network error]";
        activeSource.close();
        btn.disabled = false;
      };
    }

    // Allow Ctrl+Enter to submit
    document.getElementById("prompt").addEventListener("keydown", function(e) {
      if (e.key === "Enter" && (e.ctrlKey || e.metaKey)) startStream();
    });
  </script>
</body>
</html>

Starting the Web Server

Open http://localhost:8000. Type a prompt, click Send, and watch tokens appear character by character. The browser console’s Network tab will show the /stream request with text/event-stream content type and a steadily growing response body.

Sample Web Stream Output

# Raw SSE frames received by the browser (simplified):
data: {"token": "HTTP"}
data: {"token": " chunk"}
data: {"token": "ed"}
data: {"token": " transfer"}
data: {"token": " encoding"}
data: {"token": " allows"}
...
data: {"done": true, "input_tokens": 38, "output_tokens": 94, "stop_reason": "end_turn"}

Architecture Walkthrough: Sync SDK Inside an Async Framework

The Anthropic Python SDK’s client.messages.stream() is a synchronous context manager. FastAPI runs on an async event loop (asyncio). If you call the sync stream directly inside an async def route, you block the event loop and no other requests can be served while Claude is generating.

The pattern used above bridges this gap cleanly:

1. An asyncio.Queue sits between the sync producer and the async consumer.
2. The sync SDK runs inside loop.run_in_executor(None, _run_stream), which places it in Python’s default thread pool, leaving the event loop free.
3. Each token is pushed into the queue with loop.call_soon_threadsafe(queue.put_nowait, item), which is the correct way to communicate from a background thread to an asyncio queue.
4. The async generator in the route handler calls await queue.get() and immediately yields the SSE frame. The event loop is free between awaits.

Alternative: httpx-based Async Stream

Anthropic’s SDK also ships an async client (anthropic.AsyncAnthropic) which you can use if you want a fully async path without thread pools. The tradeoff is that AsyncAnthropic requires an async with block and you must keep the client alive for the lifespan of your FastAPI app. The sync + executor pattern above is slightly simpler for most setups and works fine under moderate concurrent load (100-200 simultaneous streams on a modern server).

Common Pitfalls and How to Avoid Them

• Forgetting flush=True in CLI scripts. Python’s stdout is line-buffered by default when connected to a terminal and block-buffered when piped. Without flush=True on each print(text, end="") call, tokens accumulate in the OS buffer and arrive in batches, defeating the purpose of streaming.
• Blocking the event loop. Calling the sync client.messages.stream() directly inside an async def function (without run_in_executor) makes FastAPI unable to handle any other request during generation. Under load, every concurrent user waits in a queue behind the current stream. Use the thread-pool bridge shown above.
• Missing X-Accel-Buffering: no header behind nginx. Nginx buffers proxied responses by default. Without this header, the browser gets tokens in large batches every few seconds instead of seeing them one by one. Add proxy_buffering off; in the nginx config or pass X-Accel-Buffering: no in the FastAPI response headers.
• Not handling stream.get_final_message() after the loop. If you exit the with block before calling get_final_message(), you lose the output token count from the terminal message_delta event. Call it once the text_stream loop is exhausted; it is instant at that point since the data is already buffered.
• Holding the with stream: context open across an await. The sync stream context manager holds an open HTTP connection. If you try to call await something() inside the with client.messages.stream() as stream: block (in a thread), the GIL and connection lifecycle interact unpredictably. Keep the sync stream in its own non-async function and bridge with a queue or executor.
• Sending the entire accumulated text as a single SSE frame. Some implementations concatenate tokens in a buffer and send them as one chunk. This works but defeats the purpose. Send each delta as its own data: line. Clients handle rapid small frames well.
• Using EventSource for prompts that need POST semantics. The browser EventSource API only supports GET requests. If your prompt is long (thousands of characters) or includes sensitive data that should not appear in server access logs, you need a different approach: open a WebSocket, or use fetch with ReadableStream on the client side (supported in all modern browsers).

Cost and Latency Considerations

Streaming does not change how many tokens you use or what you are billed. The token count for a streamed call is identical to the same call made without streaming. What changes is when the tokens are delivered to the client. Cost is determined entirely by input tokens plus output tokens, regardless of the streaming flag.

First-token latency is the metric that matters most for perceived UX. Users tolerate 100 ms before they start to notice a delay. At 200 ms they notice but accept it. At 500 ms or more they start to wonder if something is wrong. Haiku’s 60-100 ms first-token latency makes it the natural pick for any chat interface where speed matters more than depth of reasoning.

If you are building a product where token cost is a concern, combine the streaming pattern here with the caching pattern from Part 4: Prompt Caching with Claude. A large system prompt that is streamed repeatedly can have its input tokens cached, cutting your per-request input cost by 90 percent after the first call.

For routing decisions, where you send simple queries to Haiku and complex ones to Sonnet or Opus, see Part 27: Cut AI Costs: Model Routing and Batching with Claude.

Deploying to Production

Gunicorn + Uvicorn Workers

In production, run uvicorn workers under gunicorn for process supervision:

Each worker handles concurrent streams independently. Four workers on a 2-core server can comfortably support 50 to 100 simultaneous streams without latency degradation, because the work is I/O-bound (waiting on Anthropic’s API).

Nginx Reverse Proxy Config

location /stream {
    proxy_pass         http://127.0.0.1:8000;
    proxy_http_version 1.1;
    proxy_set_header   Connection "";
    proxy_buffering    off;          # critical for SSE
    proxy_cache        off;
    proxy_read_timeout 120s;         # longer than max generation time
}

The proxy_buffering off directive is the single most common production fix for SSE. Without it, nginx accumulates the entire response before forwarding it, exactly cancelling the benefit of streaming.

Timeouts

Set proxy_read_timeout to longer than your longest expected generation. A 4096-token response at Sonnet’s generation speed takes roughly 20-40 seconds. 120 seconds gives generous headroom. If your upstream load balancer (ALB, Cloudflare, etc.) has its own idle-connection timeout, raise it to at least 90 seconds for SSE endpoints.

Connection Limit and Rate Limiting

Each open SSE stream holds one server-side connection and one Anthropic API connection. If you have 1000 concurrent users, you need 1000 outbound connections to Anthropic. Check your Anthropic API rate limits (requests per minute and tokens per minute for your tier) and plan capacity accordingly. A simple token bucket rate limiter per user (using Redis + slowapi for FastAPI) protects you from a single user exhausting your rate limit.

Integrating with a React Frontend

If you prefer a React app over the plain HTML demo, the EventSource pattern is the same. Here is the core hook:

// useClaudeStream.ts (React hook, TypeScript)
import { useState, useRef, useCallback } from "react";

export function useClaudeStream() {
  const [text, setText]     = useState("");
  const [loading, setLoading] = useState(false);
  const [tokens, setTokens] = useState<{input:number,output:number} | null>(null);
  const sourceRef = useRef<EventSource | null>(null);

  const startStream = useCallback((prompt: string) => {
    if (sourceRef.current) sourceRef.current.close();
    setText("");
    setTokens(null);
    setLoading(true);

    const url = `/stream?prompt=${encodeURIComponent(prompt)}`;
    const source = new EventSource(url);
    sourceRef.current = source;

    source.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.token !== undefined) {
        setText((prev) => prev + data.token);
      }
      if (data.done) {
        setTokens({ input: data.input_tokens, output: data.output_tokens });
        setLoading(false);
        source.close();
      }
      if (data.error) {
        setText((prev) => prev + `\n[Error]: ${data.error}`);
        setLoading(false);
        source.close();
      }
    };

    source.onerror = () => {
      setLoading(false);
      source.close();
    };
  }, []);

  return { text, loading, tokens, startStream };
}

This hook encapsulates all stream lifecycle management. Your component calls startStream(prompt) and reads text for real-time display. For related patterns in multi-turn conversations, see Part 13: Build a Customer Support Agent with Claude.

Frequently Asked Questions

Does streaming affect the quality or content of Claude’s response?

No. The model generates tokens identically whether you use streaming or a blocking call. You are only changing when the tokens are delivered to your client, not how they are produced. The final response will be the same either way.

Can I cancel a stream mid-way?

Yes. Close the EventSource on the client side (call source.close()). On the server side, FastAPI will detect the disconnection and the async generator will stop yielding. The with client.messages.stream() context manager will exit cleanly when the generator function’s garbage collection runs, which closes the underlying HTTP connection to Anthropic and stops billing for further generation. There is a brief window (sub-second) where a few more tokens may be generated after the client disconnects, but you will not be billed for tokens after the connection closes at Anthropic’s end.

Is SSE the right transport, or should I use WebSockets?

SSE is simpler for one-directional server-to-client streaming. It uses plain HTTP, works through proxies and load balancers without special config (aside from buffering), and is natively supported in every modern browser. Use WebSockets if you need bidirectional, real-time communication, for example if the user can interrupt a generation or send new input while Claude is still writing. For most chat and streaming text UIs, SSE is sufficient and easier to deploy.

How do I handle multiple concurrent users?

Each user gets their own SSE connection and their own Claude API call. FastAPI with uvicorn workers handles this naturally. The bottleneck is usually your Anthropic rate limit (requests per minute) rather than server CPU. Implement per-user rate limiting and queue long-running prompts if you expect bursts. The thread-pool bridge pattern in this article scales well because threads are cheap compared to the I/O wait time on the Anthropic API.

What happens if the network drops mid-stream?

The browser’s EventSource will automatically attempt to reconnect after a brief delay. When it reconnects, your server starts a brand new Claude API call. If you need to resume from where the stream stopped (to avoid re-billing input tokens and to avoid showing a duplicate partial response), you need to implement a session ID and server-side buffer that can replay the accumulated text on reconnect. For most applications this complexity is not worth it; showing a fresh response on reconnect is acceptable.

Can I use streaming with tool use or structured output?

Yes, but the user-facing incremental text benefit does not apply to tool use calls, since tool inputs are JSON and you typically need the complete JSON before you can act on it. The SDK will still stream the tool-use response, but you would buffer the entire tool input block before parsing it. The streaming pattern in this article is most valuable for text generation. For tool use patterns, see Part 2: Tool Use with Claude.

How do I measure first-token latency in production?

Record a timestamp immediately before opening the with client.messages.stream(): block, and record another inside the for text in stream.text_stream: loop on the very first iteration. The difference is your first-token latency. Emit this as a metric (Prometheus histogram or a structured log line) on every request. Tracking p50/p95/p99 of first-token latency over time will tell you whether API latency is degrading before your users file support tickets. For broader observability patterns, see Part 28: Observability for LLM Apps: Trace and Debug Claude in Production.

Back to AI in Production: 30 Real-World Use Cases with Claude.

Further Reading

• Anthropic API: Messages Streaming Reference
• Anthropic Model Overview: Choosing the Right Tier
• FastAPI: StreamingResponse Documentation
• MDN: Using Server-Sent Events
• Anthropic Pricing