Stream Async Text

Async Streaming Recap

Lesson 56 introduced async streaming: the same stream helpers as the sync client, used with async with, async for, and await:

async with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Explain Python generators in three sentences."}],
) as stream:
    async for text in stream.text_stream:
        print(text, end="", flush=True)
    message = await stream.get_final_message()

This lesson builds on that foundation with patterns that only async makes practical:

• wrapping streams in async generators that other code can consume,
• iterating events asynchronously,
• running several streams at the same time, and
• merging concurrent streams into one sequence of chunks (fan-in).

Async Generators: Reusable Streaming Functions

An async generator is an async def function that contains yield. It produces values over time, and callers consume them with async for. Wrapping a Claude stream in one hides the SDK details from the rest of your code:

from collections.abc import AsyncIterator

from anthropic import AsyncAnthropic


async def stream_text(client: AsyncAnthropic, prompt: str) -> AsyncIterator[str]:
    async with client.messages.stream(
        model="claude-opus-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for text in stream.text_stream:
            yield text

async for chunk in stream_text(client, "Explain closures in two sentences."):
    print(chunk, end="", flush=True)

The caller sees only text chunks. The stream is opened and closed inside the generator.

Closing Async Generators Promptly

If the caller stops early, for example with break, the async generator is not closed immediately. It is closed later, when it is garbage-collected, and until then the underlying stream may stay open. Use contextlib.aclosing() (Python 3.10+) to close it as soon as the loop ends:

from contextlib import aclosing

async with aclosing(stream_text(client, "List 50 Python tips.")) as chunks:
    async for chunk in chunks:
        if "10." in chunk:
            break
# The generator, and the Claude stream inside it, are closed here

aclosing() calls the generator's aclose() method when the block ends, which runs the generator's cleanup, including the inner async with that closes the stream (Lesson 60).

Iterating Events Asynchronously

Events (Lessons 53 and 54) work the same way with async for:

async with client.messages.stream(model="claude-opus-5-5", max_tokens=2048, messages=messages) as stream:
    async for event in stream:
        if event.type == "content_block_start" and event.content_block.type == "thinking":
            print("(thinking...)")
        elif event.type == "text":
            print(event.text, end="", flush=True)
    final = await stream.get_final_message()

The helper events (text, input_json, content_block_stop, …), snapshots, and current_message_snapshot are all available.

Several Streams at Once

Because async streams do not block each other, several can run concurrently. Use cases include:

Running streams concurrently is simple with asyncio.gather(). The interesting part is displaying or combining their output while they run.

Fan-In: Merging Concurrent Streams

Fan-in means combining several concurrent sources into a single sequence. With an asyncio.Queue, each stream's task puts tagged chunks into one queue, and a single consumer reads them in the order they arrive:

stream "beginner" ──┐
stream "expert"   ──┼──► queue of (name, chunk) ──► one consumer (display, save, forward)
stream "analogy"  ──┘

This keeps the consumer simple. It handles one chunk at a time, whichever stream it came from, and is useful for terminals, logs, and forwarding several streams over a single WebSocket connection.

Practical Example: Three Explanations Streaming at Once

This program asks for three explanations of the same topic, for a beginner, for an expert, and with an analogy, concurrently. While they stream, it shows a live progress line with the characters received per stream. When all three finish, it prints the complete texts.

"""stream_fanin.py: stream several answers concurrently and merge their chunks."""

import asyncio
from collections.abc import AsyncIterator

import anthropic
from anthropic import AsyncAnthropic

PROMPTS = {
    "beginner": "Explain Python decorators to a complete beginner in about 60 words.",
    "expert": "Explain Python decorators to an experienced developer in about 60 words.",
    "analogy": "Explain Python decorators using a real-world analogy, in about 60 words.",
}


async def stream_text(client: AsyncAnthropic, prompt: str) -> AsyncIterator[str]:
    async with client.messages.stream(
        model="claude-opus-5-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for text in stream.text_stream:
            yield text


async def pump(name: str, client: AsyncAnthropic, prompt: str, queue: asyncio.Queue) -> None:
    """Copy one stream's chunks into the shared queue, then signal the end with None."""
    try:
        async for text in stream_text(client, prompt):
            await queue.put((name, text))
    except anthropic.APIError as error:
        await queue.put((name, f"\n[error: {type(error).__name__}]"))
    finally:
        await queue.put((name, None))


async def main() -> dict[str, str]:
    queue: asyncio.Queue = asyncio.Queue()
    buffers: dict[str, list[str]] = {name: [] for name in PROMPTS}

    async with AsyncAnthropic() as client:
        tasks = [asyncio.create_task(pump(name, client, prompt, queue))
                 for name, prompt in PROMPTS.items()]

        remaining = len(tasks)
        while remaining:
            name, text = await queue.get()
            if text is None:
                remaining -= 1
                print(f"\n[{name} finished]")
                continue
            buffers[name].append(text)
            sizes = " | ".join(f"{n}: {sum(map(len, parts)):>4} chars" for n, parts in buffers.items())
            print(f"\r{sizes}", end="", flush=True)

        await asyncio.gather(*tasks)

    return {name: "".join(parts).strip() for name, parts in buffers.items()}


if __name__ == "__main__":
    for name, text in asyncio.run(main()).items():
        print(f"\n=== {name} ===\n{text}")

Example output (shortened):

beginner:  212 chars | expert:  187 chars | analogy:  240 chars
[expert finished]
beginner:  356 chars | expert:  402 chars | analogy:  351 chars
[beginner finished]
beginner:  371 chars | expert:  402 chars | analogy:  396 chars
[analogy finished]

=== beginner ===
A decorator is a function that wraps another function to add extra behavior...

=== expert ===
Decorators are callables that take a function and return a replacement, applied with @ syntax...

=== analogy ===
Think of a decorator like gift wrapping: the gift (your function) stays the same...

How It Works

1. stream_text() is an async generator that wraps one Claude stream and yields only text.
2. pump() runs one stream and copies each chunk into the shared queue as (name, chunk), tagged with the stream's name. In finally, it always puts (name, None), an end marker, even if the stream failed, so the consumer never waits forever.
3. API errors inside one stream are turned into a visible error chunk for that stream only. The other streams continue.
4. main() starts one pump task per prompt with asyncio.create_task(), so all three streams run at the same time.
5. The consumer loop reads (name, chunk) pairs as they arrive, from whichever stream produced them. It counts end markers in remaining and stops when every stream has finished.
6. The progress line sums the lengths of each stream's chunks with sum(map(len, parts)), and rewrites the line with \r. Each "finished" message starts on a new line.
7. await asyncio.gather(*tasks) waits for the tasks to finish completely, which surfaces any unexpected exception from a task, before the client is closed.

The total time is roughly that of the slowest of the three streams, not their sum.

Common Mistakes

Running Concurrent Streams in Sequence

What happens: for prompt in prompts: async with client.messages.stream(...) streams one answer after another.

Fix: start each stream in its own task, and merge their output.

Forgetting End Markers in Fan-In

What happens: the consumer waits forever for a stream that failed before sending its last chunk.

Fix: always send an end marker, in finally.

Breaking Out of Async Generators Without Closing Them

What happens: the underlying stream stays open until garbage collection.

Fix: use contextlib.aclosing().

Mixing Output from Concurrent Streams Without Tags

What happens: chunks from different streams are interleaved into unreadable text.

Fix: tag each chunk with its source, and buffer per stream.

Using for Instead of async for

What happens: TypeError, because async streams and generators are not regular iterables.

Fix: use async for for async streams and async generators.

Best Practices

• Wrap streams in async generators to give the rest of your code a simple interface.
• Close async generators with aclosing() when you may stop early.
• Run independent streams in separate tasks.
• Merge concurrent streams through a queue, with tagged chunks and end markers.
• Isolate errors per stream, so one failure does not stop the others.
• Await all tasks before closing the client.

Exercise

1. Add a fourth prompt to PROMPTS and confirm that the total time hardly changes.
2. Modify pump() to also put the final output token count for each stream into the queue, and print it when the stream finishes.
3. Use aclosing() to stop the beginner stream after 150 characters, while the others continue.
4. Explain why pump() puts its end marker in a finally block instead of after the async for loop.

Step 2: have the generator yield a final usage marker, or open the stream directly in pump():

async def pump(name, client, prompt, queue):
    tokens = 0
    try:
        async with client.messages.stream(
            model="claude-opus-5-5", max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        ) as stream:
            async for text in stream.text_stream:
                await queue.put((name, text))
            tokens = (await stream.get_final_message()).usage.output_tokens
    except anthropic.APIError as error:
        await queue.put((name, f"\n[error: {type(error).__name__}]"))
    finally:
        await queue.put((name, None))
        print(f"\n[{name}: {tokens} output tokens]")

Step 4: if the stream raises an exception that is not an APIError, or if the task is cancelled, the code after the loop would never run, and the consumer would wait forever for that stream's end marker. A finally block runs in every case, so the consumer is always told that the stream is done.