Stream Async Responses

Why Async Streaming?

The streaming examples so far used the synchronous client, Anthropic. While a synchronous stream is running, the code that reads it is blocked: it waits for each chunk before doing anything else. That is fine for scripts and command-line tools.

In asynchronous programs, such as FastAPI web servers, Discord bots, or tools that handle several conversations at once, blocking is a serious problem. A blocked event loop cannot serve other users while one stream is in progress.

The asynchronous client, AsyncAnthropic, streams without blocking. While waiting for the next chunk, the event loop can work on other tasks, such as other users' streams.

import asyncio

from anthropic import AsyncAnthropic

client = AsyncAnthropic()


async def main() -> None:
    async with client.messages.stream(
        model="claude-opus-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Write a haiku about async code."}],
    ) as stream:
        async for text in stream.text_stream:
            print(text, end="", flush=True)
        message = await stream.get_final_message()

    print(f"\n[{message.usage.output_tokens} output tokens]")


asyncio.run(main())

The Async SDK section (Lessons 71–80) covers async programming fully. This lesson focuses on async streaming.

Sync vs. Async Streaming

The async API mirrors the sync API. Only the keywords change:

Why these keywords:

• async with opens and closes the stream asynchronously, so even connection setup and cleanup do not block.
• async for waits for each chunk cooperatively. While no chunk is available, the event loop runs other tasks.
• await waits for the final message without blocking other tasks.

Everything you learned about events (Lesson 53), helpers (Lesson 54), and final messages (Lesson 55) applies unchanged.

Streaming Several Responses Concurrently

Because async streams do not block each other, several streams can run at the same time in one program:

import asyncio
import time

from anthropic import AsyncAnthropic

client = AsyncAnthropic()


async def stream_one(label: str, question: str) -> tuple[str, int]:
    chunks = []
    async with client.messages.stream(
        model="claude-opus-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": question}],
    ) as stream:
        async for text in stream.text_stream:
            chunks.append(text)
        message = await stream.get_final_message()
    return label, message.usage.output_tokens


async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        stream_one("lists", "Explain Python lists in 3 sentences."),
        stream_one("dicts", "Explain Python dictionaries in 3 sentences."),
        stream_one("sets", "Explain Python sets in 3 sentences."),
    )
    print(results)
    print(f"All three finished in {time.perf_counter() - start:.1f}s")


asyncio.run(main())

How it works:

• asyncio.gather(...) runs the three coroutines concurrently and waits for all of them. Results are returned in the same order as the arguments.
• While one stream waits for its next chunk, the others make progress, so the total time is close to the time of the slowest single request, not the sum of all three.

Running many requests concurrently can hit rate limits. Lesson 75 shows how to limit concurrency with a semaphore.

Streaming to a Web Browser with FastAPI

A common use of async streaming is a web endpoint that forwards Claude's text to a browser as it arrives. FastAPI's StreamingResponse sends the body incrementally from an async generator.

The browser can receive the stream as server-sent events (SSE), the same format the Claude API uses (Lesson 51). Each SSE message is a line starting with data:, followed by a blank line:

data: {"text": "Hello"}

data: {"text": " world"}

Requirements

python -m pip install fastapi uvicorn

Practical Example: A FastAPI Streaming Endpoint

This small web application streams Claude's answer to the browser using SSE, with a minimal HTML page that displays the text as it arrives.

app.py

"""app.py: stream Claude's answers to a browser with FastAPI and server-sent events."""

import json

import anthropic
from anthropic import AsyncAnthropic
from fastapi import FastAPI, Query
from fastapi.responses import HTMLResponse, StreamingResponse

app = FastAPI()
client = AsyncAnthropic()   # One client for the whole application


def sse(data: dict) -> str:
    """Format one server-sent event."""
    return f"data: {json.dumps(data)}\n\n"


async def answer_events(question: str):
    try:
        async with client.messages.stream(
            model="claude-opus-5-5",
            max_tokens=2048,
            system="You are a concise Python tutor. Answer in under 150 words.",
            messages=[{"role": "user", "content": question}],
        ) as stream:
            async for text in stream.text_stream:
                yield sse({"type": "text", "text": text})
            final = await stream.get_final_message()
        yield sse({"type": "done", "stop_reason": final.stop_reason})
    except anthropic.APIError:
        yield sse({"type": "error", "message": "The answer could not be completed. Please try again."})


@app.get("/ask")
async def ask(q: str = Query(..., min_length=1, max_length=2000)):
    return StreamingResponse(answer_events(q), media_type="text/event-stream")


PAGE = """<!doctype html>
<html>
<body>
  <input id="q" size="60" placeholder="Ask a Python question">
  <button onclick="ask()">Ask</button>
  <pre id="out" style="white-space: pre-wrap"></pre>
  <script>
    function ask() {
      const out = document.getElementById("out");
      out.textContent = "";
      const source = new EventSource("/ask?q=" + encodeURIComponent(document.getElementById("q").value));
      source.onmessage = (event) => {
        const data = JSON.parse(event.data);
        if (data.type === "text") out.textContent += data.text;
        else { if (data.type === "error") out.textContent += "\\n[" + data.message + "]"; source.close(); }
      };
      source.onerror = () => source.close();
    }
  </script>
</body>
</html>"""


@app.get("/", response_class=HTMLResponse)
async def index():
    return PAGE

Run the server and open http://127.0.0.1:8000 in a browser:

uvicorn app:app --reload

How It Works

1. AsyncAnthropic() is created once at module level and shared by all requests, so connections are reused (Lesson 10).
2. answer_events() is an async generator: a function with async with, async for, and yield. Each yield sends one chunk of the HTTP response to the browser immediately.
3. sse() formats each chunk as a server-sent event. The text is JSON-encoded, because raw text can contain line breaks, which would break the SSE format, where a blank line ends an event.
4. Event types (text, done, error) let the browser distinguish answer text from the end of the stream and from errors.
5. Errors are sent as events, with a generic message, rather than crashing the response. Internal error details stay on the server. Because the HTTP response has already started, an error status code can no longer be sent; an error event is the correct way to report problems mid-stream.
6. Query(..., min_length=1, max_length=2000) validates the question: it is required, not empty, and limited in size (Lesson 18).
7. EventSource in the browser opens the SSE connection, receives each event in onmessage, and appends text to the page. The browser closes the connection on done or error. Otherwise EventSource would automatically reconnect and ask the question again.
8. While one user's answer is streaming, the server can serve other users, because the async stream never blocks the event loop.

Note: EventSource only supports GET requests, so the question is sent in the URL. For long inputs or conversation history, use POST with fetch() and read the response body as a stream. Lesson 66 builds a full web chatbot.

Common Mistakes

Using the Sync Client in Async Code

What happens: Anthropic().messages.stream(...) inside an async def endpoint blocks the event loop, so every other request on the server waits.

Fix: use AsyncAnthropic with async with and async for.

Forgetting async Keywords

What happens: with client.messages.stream(...) with an async client raises an error, and stream.get_final_message() without await returns a coroutine instead of a message.

Fix: use async with, async for, and await consistently.

Sending Raw Text as SSE Data

What happens: line breaks inside the text break the event format, and the browser receives fragments or empty events.

Fix: JSON-encode each chunk.

Letting the Browser Reconnect Forever

What happens: EventSource reconnects automatically after the stream ends, and the question is asked again and again.

Fix: send a final done event, and call source.close() in the browser.

Creating a New Client per Request

What happens: each request opens new connections, adding latency.

Fix: create one AsyncAnthropic client for the application.

Best Practices

• Use AsyncAnthropic in async frameworks and concurrent programs.
• Mirror the sync patterns with async with, async for, and await.
• Run independent streams concurrently with asyncio.gather, while respecting rate limits.
• Forward streams to browsers as JSON-encoded SSE events, with explicit done and error events.
• Validate inputs before starting a stream.
• Share one async client across the application.

Exercise

1. Run the concurrency example with three questions, then rewrite it to run the same three requests one after another (without gather). Compare the total times.
2. Extend answer_events() to send a "thinking" event when a thinking block starts, by iterating events instead of text_stream.
3. Add the output_tokens value to the done event, and display it in the browser.
4. Explain why the endpoint cannot return an HTTP 500 status if an error happens halfway through the stream.

Step 2: iterate events and handle the thinking block start:

async with client.messages.stream(...) as stream:
    async for event in stream:
        if event.type == "content_block_start" and event.content_block.type == "thinking":
            yield sse({"type": "thinking"})
        elif event.type == "text":
            yield sse({"type": "text", "text": event.text})
    final = await stream.get_final_message()

In the browser, show a "Thinking…" message when data.type === "thinking", and clear it when text arrives.

Step 4: the HTTP status code and headers are sent before the first chunk of the body. Once streaming has started, the status (200) has already been delivered to the browser and cannot be changed. Errors that happen later must be reported inside the stream, which is why the example sends an error event.