Build a Live Chat

What a Live Chat Is

A live chat is a chat loop (Lesson 26) in which Claude's answers stream onto the screen as they are generated, the experience users know from chat applications. It combines everything from the Messages and Streaming sections:

This lesson builds a complete terminal live chat step by step, explaining each design decision.

The Core Loop

Compared with the non-streaming chat loop in Lesson 26, one part changes: instead of create() and printing the full answer, each turn streams the answer and then retrieves the final message for the history.

read user input
   ↓
build pending history (history + new user message)
   ↓
stream the reply: show status and text live
   ↓
get the final message
   ↓
check the stop reason → commit the turn to history (or not)
   ↓
repeat

The final message is what goes into the history, never the streamed text (Lesson 55). This keeps thinking blocks and any tool calls intact.

Design Decisions

1. Show Something Immediately

With thinking models, the first answer text can take a moment. Printing Claude: right away, and (thinking...) when a thinking block starts, tells the user that work is in progress.

2. Let Users Stop an Answer

Users often realize mid-answer that they asked the wrong question. In a terminal, Ctrl+C raises KeyboardInterrupt. Catching it during the stream stops the answer without ending the whole program. Raising the exception inside the with block closes the stream (Lesson 60).

What to do with a stopped answer: this chat does not store it, because a partial answer is not a complete turn, and the user can simply ask again. An alternative design stores the partial text as a synthetic assistant turn, marked as interrupted.

3. Commit Only Complete, Usable Turns

As in Lessons 24 and 26, the user message is added to a pending copy of the history. The history is updated only when the stream completes successfully and the reply is not a refusal.

4. Handle Errors Per Turn

A network problem or a mid-stream error ends that answer, not the chat. The user sees a short message and can try again.

5. Keep the History Bounded

Long chats are trimmed to a maximum number of messages, so each turn stays fast and affordable (Lesson 24).

Practical Example: A Complete Terminal Live Chat

"""live_chat.py: a streaming terminal chat with Claude.

Commands: /retry  /reset  /tokens  /quit
Press Ctrl+C while Claude is answering to stop the answer.
"""

import anthropic
from anthropic import Anthropic
from anthropic.types import Message

MODEL = "claude-opus-5-5"
SYSTEM = "You are a friendly programming tutor. Keep answers under 150 words unless asked for more."
MAX_HISTORY_MESSAGES = 30

client = Anthropic()


def trim(messages: list[dict]) -> list[dict]:
    trimmed = messages[-MAX_HISTORY_MESSAGES:]
    while trimmed and trimmed[0]["role"] != "user":
        trimmed = trimmed[1:]
    return trimmed


def stream_reply(messages: list[dict]) -> Message | None:
    """Stream one reply to the terminal. Return the final message, or None if it did not complete."""
    thinking_shown = False
    try:
        with client.messages.stream(
            model=MODEL, max_tokens=2048, system=SYSTEM, messages=messages
        ) as stream:
            print("Claude: ", end="", flush=True)
            for event in stream:
                if (event.type == "content_block_start"
                        and event.content_block.type == "thinking"
                        and not thinking_shown):
                    print("(thinking...) ", end="", flush=True)
                    thinking_shown = True
                elif event.type == "text":
                    print(event.text, end="", flush=True)
            final = stream.get_final_message()
    except KeyboardInterrupt:
        print("\n[Answer stopped. It was not saved.]")
        return None
    except anthropic.APIError as error:
        print(f"\n[The answer failed ({type(error).__name__}). Please try again.]")
        return None

    print()
    return final


def main() -> None:
    history: list[dict] = []
    total_input = total_output = 0
    print("Live chat with Claude. Commands: /retry /reset /tokens /quit")
    print("Press Ctrl+C during an answer to stop it.\n")

    while True:
        try:
            user_text = input("You: ").strip()
        except (KeyboardInterrupt, EOFError):
            break

        if not user_text:
            continue
        if user_text in {"/quit", "quit", "exit"}:
            break
        if user_text == "/reset":
            history = []
            print("[New conversation started.]\n")
            continue
        if user_text == "/tokens":
            print(f"[Tokens this session: input={total_input}, output={total_output}]\n")
            continue

        if user_text == "/retry":
            if len(history) < 2:
                print("[Nothing to retry.]\n")
                continue
            pending = history[:-1]                       # Drop the last answer; ends with the user message
        else:
            pending = trim(history + [{"role": "user", "content": user_text}])

        final = stream_reply(pending)
        if final is None:
            print()
            continue

        total_input += final.usage.input_tokens
        total_output += final.usage.output_tokens

        if final.stop_reason == "refusal":
            print("[Claude declined this request. It was not saved.]\n")
            continue
        if final.stop_reason == "max_tokens":
            print("[Note: the answer was cut off. Ask Claude to continue.]")

        history = pending + [{"role": "assistant", "content": final.to_dict()["content"]}]
        print()

    print(f"\nGoodbye! This session used {total_input} input and {total_output} output tokens.")


if __name__ == "__main__":
    main()

Example session:

Live chat with Claude. Commands: /retry /reset /tokens /quit
Press Ctrl+C during an answer to stop it.

You: What is a Python generator?
Claude: (thinking...) A generator is a function that produces values lazily using yield...

You: Show a tiny example.
Claude: Here's one:

def count_up(limit):
    for n in range(limit):
        yield n
...

You: Write a 2,000-word history of programming languages.
Claude: Programming languages began with...^C
[Answer stopped. It was not saved.]

You: /tokens
[Tokens this session: input=731, output=402]

You: /quit

Goodbye! This session used 731 input and 402 output tokens.

How It Works

1. stream_reply() handles one streamed answer. It iterates events (Lesson 54) rather than text_stream, so it can react to a thinking block and print text. thinking_shown makes sure the indicator appears only once per answer.
2. get_final_message() is called inside the with block after the loop and returns the complete Message, which is what goes into the history.
3. except KeyboardInterrupt catches Ctrl+C during the answer. Because the exception is raised inside the with block, the stream is closed before the handler runs. The chat continues, and the stopped answer is not stored.
4. except anthropic.APIError catches API errors and errors raised mid-stream. The chat continues, and the history is unchanged.
5. pending is always a new list. history is only replaced after a complete, non-refused answer, so failures, stops, and refusals never leave a dangling user message.
6. /retry sends the history without the last answer. The new answer replaces the old one, because the history is rebuilt from pending.
7. Token totals count every completed answer, including refusals, because those tokens are billed. Stopped and failed answers are not counted here, because their usage is not available. Lesson 60 discusses this trade-off.
8. final.to_dict()["content"] stores the complete assistant turn as plain data, ready to be saved to JSON (Lesson 24).

Extending the Live Chat

Common Mistakes

Storing Streamed Text Instead of the Final Message

What happens: thinking blocks and tool calls are lost, and later requests may fail or lose context.

Fix: store final.to_dict()["content"].

Letting Ctrl+C End the Whole Program

What happens: stopping one answer quits the chat and loses the conversation.

Fix: catch KeyboardInterrupt around the stream separately from the input prompt.

Committing the User Message Before the Answer Completes

What happens: after a failure or a stop, the next request contains two consecutive user messages.

Fix: use a pending copy, and commit only after success.

No Feedback Before the First Text

What happens: users think the chat has frozen during thinking.

Fix: print a label immediately, and show a thinking indicator.

Unbounded History

What happens: each turn becomes slower and more expensive.

Fix: trim or summarize the history.

Best Practices

• Stream every answer, and show feedback immediately.
• Store final messages, not streamed text.
• Let users stop answers without leaving the chat.
• Commit turns only after successful, usable answers.
• Handle errors per turn, with friendly messages.
• Bound the history, and track token usage.
• Keep the stream logic in its own function, so it can be reused in other interfaces.

Exercise

1. Add /save and /load commands that write and read the history as JSON.
2. Change the stop behavior so that a stopped answer is stored as a text-only assistant turn ending with " [stopped by user]". Hint: collect the text from text events in a list.
3. Show the time to first text after each answer, for example [first text after 1.4s].
4. Explain why stream_reply() iterates events instead of using text_stream.

Step 2: collect the text as it streams, and return it when the answer is stopped:

def stream_reply(messages):
    shown, chunks = False, []
    try:
        with client.messages.stream(model=MODEL, max_tokens=2048, system=SYSTEM, messages=messages) as stream:
            print("Claude: ", end="", flush=True)
            for event in stream:
                if event.type == "content_block_start" and event.content_block.type == "thinking" and not shown:
                    print("(thinking...) ", end="", flush=True)
                    shown = True
                elif event.type == "text":
                    chunks.append(event.text)
                    print(event.text, end="", flush=True)
            return stream.get_final_message(), None
    except KeyboardInterrupt:
        print("\n[Answer stopped.]")
        partial = "".join(chunks).strip()
        return None, (partial + " [stopped by user]") if partial else None

In main(), if the function returns stopped text, commit it as {"role": "assistant", "content": stopped_text}. A text-only assistant turn is allowed (Lesson 23), so the conversation stays valid and Claude can see where it was interrupted.

Step 4: text_stream yields only answer text, so the chat could not detect the start of a thinking block. Iterating events gives access to both the content_block_start event (for the thinking indicator) and the text helper events (for the answer). A stream should be consumed in only one way (Lesson 54).