Build a Streaming Chat App

The Goal

A streaming chat app shows the answer as the model writes it, a few words at a time, the way people expect from modern chat products. Waiting for a full answer before showing anything feels slow even when the total time is the same, because the first words arrive in a fraction of a second. Streaming changes the experience without changing the model's answer, and it is the pattern behind most chat interfaces.

This project builds the part of the app that turns the model's stream into something a web page can display. The design keeps the streaming logic separate from the web framework: the core is a generator that yields text as it arrives, and a web layer forwards those pieces to the browser. The core can be tested with a stand-in stream, and it does not depend on any particular web framework.

What You Will Build

A ChatStreamer whose reply method is a generator, yielding each piece of text as the model produces it, and recording the final message when the stream ends. A chat_turn function adds the user's message, consumes the stream, and stores the complete reply in the history, so the next turn has the full conversation.

Project Structure

streaming_chat/
├── streamer.py    # the generator that yields text, and the turn function
└── demo.py        # a stand-in stream, and one turn consumed chunk by chunk

Step 1: The Generator

The streamer opens a stream with the client's stream helper, which is a context manager. Inside it, the text is yielded as it arrives. When the stream ends, the final message is read, and its stop reason is recorded, so the caller can tell a complete reply from a truncated one (Lesson 43):

"""streamer.py: stream a reply as a generator, so a web layer can forward each piece."""

MODEL = "claude-sonnet-5-5"


class ChatStreamer:
    def __init__(self, client):
        self.client = client
        self.stop_reason: str | None = None

    def reply(self, history: list[dict]):
        with self.client.messages.stream(model=MODEL, max_tokens=400, messages=history) as stream:
            for piece in stream.text_stream:
                yield piece
            final = stream.get_final_message()
        self.stop_reason = final.stop_reason

The with block keeps the connection open while the pieces are yielded, and closes it when the loop finishes, even if the caller stops early (Lesson 60). The stop reason is read after the stream ends, because it is known only then.

Step 2: A Turn

A turn has three parts: record the user's message, consume the stream while passing each piece to the caller, and store the full reply. The collected pieces are returned, so the caller can display them or check them:

def chat_turn(streamer: ChatStreamer, history: list[dict], user_text: str) -> tuple[list[str], str | None]:
    history.append({"role": "user", "content": user_text})
    pieces: list[str] = []
    for piece in streamer.reply(history):
        pieces.append(piece)
    reply = "".join(pieces)
    history.append({"role": "assistant", "content": reply})
    return pieces, streamer.stop_reason

In a web application, the loop body would send each piece to the browser as an event, instead of appending it to a list. The shape of the function stays the same, which is why the core is written this way.

Step 3: A Demonstration

The stand-in stream yields three pieces and a final message. Its text stream is an iterator, and it implements the context manager protocol, so it behaves as the real helper does for this code:

"""demo.py: one streamed turn, shown piece by piece, with a stand-in stream."""

from types import SimpleNamespace

from streamer import ChatStreamer, chat_turn


class FakeStream:
    def __init__(self, pieces):
        self.text_stream = iter(pieces)
        self._final = SimpleNamespace(stop_reason="end_turn",
                                      content=[SimpleNamespace(type="text", text="".join(pieces))])

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def get_final_message(self):
        return self._final


class FakeClient:
    def __init__(self):
        self.messages = self

    def stream(self, **request):
        return FakeStream(["Hel", "lo", " there."])


if __name__ == "__main__":
    history: list[dict] = []
    pieces, stop = chat_turn(ChatStreamer(FakeClient()), history, "Say hello.")
    print("chunks:", "|".join(pieces))
    print("stop:", stop)
    print("history length:", len(history), "last reply:", history[-1]["content"])

Output:

chunks: Hel|lo| there.
stop: end_turn
history length: 2 last reply: Hello there.

How It Works

1. The stream yields three pieces, which the turn collects in order. The separator in the output shows the boundaries the reader would see arriving one at a time.
2. The stop reason is read after the stream ends, and it is end_turn, meaning the reply is complete. A max_tokens stop would signal a truncated reply that the interface should mark.
3. The history gains two messages, the user's and the complete assistant reply. The stored reply is the joined text, so the next turn sees the whole answer, not the pieces.
4. The generator is consumed in the turn, which is what makes it safe to use with a web framework: the caller decides what to do with each piece.

Going Live

The stand-in is replaced by Anthropic(), and the stream helper's interface is the same. The pieces the model produces are small, often a word or part of a word, so a display that appends them directly produces smooth output.

Serving the Stream

A web framework can forward the generator's pieces as server-sent events, one event per piece, followed by a final event that carries the stop reason. The browser appends each piece to the page as it arrives. The web layer is small, because the streaming logic is already in the generator, and the framework only needs to iterate it and write each piece to the response.

The interface should handle errors that happen mid-stream. A dropped connection leaves the partial reply on screen, and the page should label it as incomplete, using the completion flag from Lesson 184, rather than leaving the user to guess.

Testing the Project

The generator and the turn function are tested with the same stand-in used here. Tests cover a complete stream, a stream that raises partway through, and a stream that stops with a truncation reason. Each test asserts on the pieces, the stored reply, and the recorded stop reason, following the approach of Lesson 184.