Streaming Terminal Chatbot with Memory

Build a command-line chatbot with the Anthropic SDK that streams replies, remembers the conversation and handles API errors.

Anthropic SDKStreamingConversation HistoryError Handling

Overview

Every chat app — ChatGPT, Claude.ai, the assistant in your IDE — is built on the same loop: send the conversation so far to a model, show the reply, add it to the conversation, repeat. In this project you build that loop yourself in about 100 lines of Python, using the official Anthropic SDK.

Your chatbot streams Claude's answer word by word (so long answers start appearing immediately), remembers earlier messages so follow-up questions work, prints how many tokens each turn used, saves transcripts on request, and recovers gracefully from network problems and rate limits instead of crashing.

What You'll Build

A client that reads your API key from the environment — never hard-coded.
Streaming output with client.messages.stream() and text_stream.
Multi-turn memory by keeping a messages list and resending it each turn.
A system prompt that gives the bot a consistent personality.
/reset, /save and /quit commands.
A typed error-handling chain for connection errors, rate limits and other API errors, plus a check for declined requests.

Prerequisites

Python 3.10 or newer.
An Anthropic API key from the Claude Console (console.anthropic.com) — lesson 3 covers creating and storing keys safely.
Provider SDK basics: client.messages.create, roles and the messages list — lesson 5.
Basic Python: functions, lists, try/except.

API Usage Costs Money

Every message you send is billed to your Anthropic account by the number of tokens processed. Because the whole conversation is resent each turn, long chats cost more per message. Set a monthly spend limit in the Console while you experiment.

Project Structure

The whole app is one file, chat.py. The ask() function is the only code that talks to the API: it streams one reply and returns the complete message object. main() owns the conversation — the messages list sent to the model and a plain-text transcript for saving — and handles commands and errors. Keeping the API call in one function makes it trivial to change the model or settings later.

Step 1: Set Up the Environment

Create a project folder with a virtual environment, install the SDK, and put your key in an environment variable. The SDK reads ANTHROPIC_API_KEY automatically, so the key never appears in your code.

mkdir claude-chat && cd claude-chat
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install anthropic

export ANTHROPIC_API_KEY="sk-ant-..."   # Windows PowerShell: $env:ANTHROPIC_API_KEY="sk-ant-..."

Step 2: Stream a Single Reply

client.messages.stream() is a context manager. Iterating over stream.text_stream yields text as soon as it is generated, and stream.get_final_message() returns the complete response afterwards — including stop_reason and token usage. output_config={"effort": "low"} asks the model to answer quickly and economically, which suits casual chat; raise it to "medium" or "high" for harder questions.

import anthropic

MODEL = "claude-opus-5-5"
SYSTEM_PROMPT = (
    "You are a friendly, concise programming tutor for beginners. "
    "Explain ideas in plain language with short code examples."
)

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from the environment


def ask(messages: list) -> anthropic.types.Message:
    """Stream Claude's reply to the terminal and return the complete message."""
    with client.messages.stream(
        model=MODEL,
        max_tokens=64000,
        system=SYSTEM_PROMPT,
        output_config={"effort": "low"},  # fast, economical answers suit casual chat
        messages=messages,
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
        print()
        return stream.get_final_message()

Step 3: Add Conversation Memory

The API is stateless — it remembers nothing between calls. "Memory" simply means appending every user message and every assistant reply to a list and sending the whole list each time. Append the reply's full content (not just its text), because responses can contain more than one kind of content block and the conversation must be sent back exactly as it was received.

messages: list = []

messages.append({"role": "user", "content": "What is a Python list?"})
reply = ask(messages)
messages.append({"role": "assistant", "content": reply.content})

messages.append({"role": "user", "content": "How is that different from a tuple?"})
reply = ask(messages)  # Claude sees the earlier question, so "that" makes sense

Step 4: Commands & Error Handling

Catch the most specific errors first. RateLimitError is a subclass of APIStatusError, so it must come before it. When a call fails, the user's last message is removed again with messages.pop(), so the history never ends with an unanswered question. A stop_reason of "refusal" means the model declined the request; that turn is dropped the same way.

try:
    reply = ask(messages)
except anthropic.APIConnectionError:
    print("\n[Network problem - check your connection and try again.]")
    messages.pop()
except anthropic.RateLimitError:
    print("\n[Rate limited - wait a moment and try again.]")
    messages.pop()
except anthropic.APIStatusError as err:
    print(f"\n[API error {err.status_code}: {err.message}]")
    messages.pop()

Complete Code

chat.py

"""chat.py - a streaming terminal chatbot with conversation memory.

Run with:  python chat.py
"""
from datetime import datetime

import anthropic

MODEL = "claude-opus-5-5"
SYSTEM_PROMPT = (
    "You are a friendly, concise programming tutor for beginners. "
    "Explain ideas in plain language with short code examples."
)

client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY from the environment


def ask(messages: list) -> anthropic.types.Message:
    """Stream Claude's reply to the terminal and return the complete message."""
    with client.messages.stream(
        model=MODEL,
        max_tokens=64000,
        system=SYSTEM_PROMPT,
        output_config={"effort": "low"},  # fast, economical answers suit casual chat
        messages=messages,
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
        print()
        return stream.get_final_message()


def save_transcript(lines: list[str]) -> str:
    path = datetime.now().strftime("chat-%Y%m%d-%H%M%S.txt")
    with open(path, "w", encoding="utf-8") as f:
        f.write("\n\n".join(lines))
    return path


def main() -> None:
    messages: list = []  # what we send to the API each turn
    transcript: list[str] = []  # plain text, for /save
    print(f"Chatting with {MODEL}. Commands: /reset  /save  /quit")

    while True:
        try:
            user_input = input("\nYou: ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break

        if not user_input:
            continue
        if user_input == "/quit":
            break
        if user_input == "/reset":
            messages.clear()
            transcript.clear()
            print("(conversation cleared)")
            continue
        if user_input == "/save":
            print(f"(saved to {save_transcript(transcript)})")
            continue

        messages.append({"role": "user", "content": user_input})
        print("Claude: ", end="", flush=True)
        try:
            reply = ask(messages)
        except anthropic.APIConnectionError:
            print("\n[Network problem - check your connection and try again.]")
            messages.pop()
            continue
        except anthropic.RateLimitError:
            print("\n[Rate limited - wait a moment and try again.]")
            messages.pop()
            continue
        except anthropic.APIStatusError as err:
            print(f"\n[API error {err.status_code}: {err.message}]")
            messages.pop()
            continue

        if reply.stop_reason == "refusal":
            print("[Claude declined to answer that request.]")
            messages.pop()
            continue

        # Keep the full content (not just the text) so the history stays valid.
        messages.append({"role": "assistant", "content": reply.content})
        text = "".join(block.text for block in reply.content if block.type == "text")
        transcript += [f"You: {user_input}", f"Claude: {text}"]
        usage = reply.usage
        print(f"  [tokens - in: {usage.input_tokens}, out: {usage.output_tokens}]")

    print("Goodbye!")


if __name__ == "__main__":
    main()

Sample Run

python chat.py (example session - replies and token counts will vary)

Click Run to see what this code prints.

Notice that the input token count grows with every turn even though each question is short — the whole conversation is sent each time. That is exactly why long chats become slower and more expensive, and why real apps eventually summarise or trim old messages.

Extend This Project

Add a /system <text> command that changes the bot's personality for the rest of the chat.
Add a /effort low|medium|high command and compare answer quality and token usage on a hard question.
Keep a running total of tokens and estimate the session cost from the model's published per-token prices.
Load a saved transcript back into messages so you can continue an old conversation.
Read the Anthropic docs on refusals and server-side fallbacks, and try routing declined requests to a fallback model automatically.

Summary

You built the core loop behind every chat product: a stateless API, conversation memory implemented as a growing messages list, a system prompt for consistent behaviour, streaming for responsiveness, and a specific-to-general error-handling chain that keeps the app alive when the network or rate limits misbehave.