Build a Chatbot

From Chat Loops to Chatbot Applications

Earlier lessons built chat loops: scripts that read input, call Claude, and print the answer (Lessons 26 and 58). Real chatbot applications need more structure:

• the same chatbot logic must work in a terminal, a web page, and an API,
• conversations must be stored and resumed (Lesson 62),
• users can have several sessions (Lesson 63), and
• the bot may remember facts about users across sessions (Lesson 64).

The key design idea for all of this is separation: one chatbot core contains the conversation logic, and each interface (CLI, Flask, FastAPI) is a thin layer on top of it.

┌──────────────┐  ┌──────────────┐  ┌──────────────┐
│ CLI (L65)    │  │ Flask (L67)  │  │ FastAPI (L68)│   ← interfaces: input/output only
└──────┬───────┘  └──────┬───────┘  └──────┬───────┘
       └────────────┬────┴─────────────────┘
             ┌──────▼───────┐
             │ Chatbot core │   ← conversation logic: history, requests,
             └──────┬───────┘     streaming, stop reasons, errors
             ┌──────▼───────┐
             │ Anthropic SDK│
             └──────────────┘

This lesson builds that core, chatbot_core.py, which the rest of the Chat Applications section reuses.

What the Core Is Responsible For

What the core does not do: printing, reading input, rendering HTML, or HTTP handling. Those belong to interfaces.

Designing the Core's Interface

The core exposes a small set of types and methods:

Why stream_events() Yields Plain Dictionaries

A generator that yields simple dictionaries, {"type": "text", "text": ...} followed by {"type": "done", "reply": ...}, works for every interface:

• a CLI prints each text event,
• a web endpoint converts each event into a server-sent event (Lessons 67 and 68), and
• the final done event carries the complete Reply, so every interface finishes the turn the same way.

Why Reply.new_messages Exists

After a successful turn, storage layers need to save exactly two new messages: the user message and the assistant message. Returning them in the Reply means storage never has to compare histories or guess what changed.

Practical Example: chatbot_core.py

"""chatbot_core.py: a reusable chatbot core for the Chat Applications lessons."""

from collections.abc import Iterator
from dataclasses import dataclass, field

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

MAX_MESSAGE_CHARS = 10_000


@dataclass
class BotConfig:
    model: str = "claude-opus-5-5"
    system: str = "You are a helpful assistant."
    max_tokens: int = 2048
    max_history_messages: int = 40


@dataclass
class Reply:
    text: str
    ok: bool
    stop_reason: str | None = None
    error: str | None = None
    input_tokens: int = 0
    output_tokens: int = 0
    new_messages: list[dict] = field(default_factory=list)


class ChatBot:
    def __init__(self, client: Anthropic, config: BotConfig | None = None,
                 history: list[dict] | None = None):
        self.client = client
        self.config = config or BotConfig()
        self.history: list[dict] = list(history or [])

    # ---------- public API ----------

    def send(self, user_text: str) -> Reply:
        problem = self._validate(user_text)
        if problem:
            return Reply("", ok=False, error=problem)

        pending = self._pending(user_text)
        try:
            final = self.client.messages.create(**self._request(pending))
        except anthropic.APIError as error:
            return Reply("", ok=False, error=type(error).__name__)
        return self._commit(pending, final)

    def stream_events(self, user_text: str) -> Iterator[dict]:
        problem = self._validate(user_text)
        if problem:
            yield {"type": "done", "reply": Reply("", ok=False, error=problem)}
            return

        pending = self._pending(user_text)
        try:
            with self.client.messages.stream(**self._request(pending)) as stream:
                for text in stream.text_stream:
                    yield {"type": "text", "text": text}
                final = stream.get_final_message()
        except anthropic.APIError as error:
            yield {"type": "done", "reply": Reply("", ok=False, error=type(error).__name__)}
            return
        yield {"type": "done", "reply": self._commit(pending, final)}

    def reset(self) -> None:
        self.history = []

    # ---------- internal helpers ----------

    def _validate(self, user_text: str) -> str | None:
        if not user_text.strip():
            return "empty_message"
        if len(user_text) > MAX_MESSAGE_CHARS:
            return "message_too_long"
        return None

    def _pending(self, user_text: str) -> list[dict]:
        messages = self.history + [{"role": "user", "content": user_text.strip()}]
        messages = messages[-self.config.max_history_messages:]
        while messages and messages[0]["role"] != "user":
            messages = messages[1:]
        return messages

    def _request(self, messages: list[dict]) -> dict:
        return {
            "model": self.config.model,
            "max_tokens": self.config.max_tokens,
            "system": self.config.system,
            "messages": messages,
        }

    def _commit(self, pending: list[dict], final: Message) -> Reply:
        text = "".join(b.text for b in final.content if b.type == "text").strip()
        usage = {"input_tokens": final.usage.input_tokens, "output_tokens": final.usage.output_tokens}

        if final.stop_reason == "refusal":
            return Reply("I can't help with that request.", ok=False,
                         stop_reason="refusal", error="refusal", **usage)

        assistant_message = {"role": "assistant", "content": final.to_dict()["content"]}
        self.history = pending + [assistant_message]
        return Reply(text, ok=True, stop_reason=final.stop_reason, **usage,
                     new_messages=[pending[-1], assistant_message])

Using the Core

"""demo_core.py: use the chatbot core with and without streaming."""

from anthropic import Anthropic

from chatbot_core import BotConfig, ChatBot

bot = ChatBot(Anthropic(), BotConfig(system="You are a concise Python tutor. Answer in under 80 words."))

reply = bot.send("My name is Priya. What is a Python tuple?")
print("Claude:", reply.text)

print("Claude: ", end="")
for event in bot.stream_events("What's my name, and how is a tuple different from a list?"):
    if event["type"] == "text":
        print(event["text"], end="", flush=True)
    else:
        final = event["reply"]
print(f"\n[ok={final.ok}, stop_reason={final.stop_reason}, history={len(bot.history)} messages]")

Example output:

Claude: Nice to meet you, Priya! A tuple is an ordered, immutable collection, written with parentheses: (1, 2, 3).
Claude: Your name is Priya. Unlike a list, a tuple can't be changed after it's created, which makes it safe for fixed data and usable as a dictionary key.
[ok=True, stop_reason=end_turn, history=4 messages]

How It Works

1. BotConfig groups all settings. Different bots (a tutor, a support agent) are just different configurations.
2. Reply is the single result type for every turn. Interfaces check ok, display text or an error, and storage saves new_messages. No interface ever handles SDK exceptions or stop reasons directly.
3. _validate() rejects empty and oversized messages before any API call, returning short error codes such as "empty_message" that interfaces can translate into friendly text (Lesson 70).
4. _pending() builds the request history: existing history plus the new message, trimmed to max_history_messages and starting with a user message (Lesson 24). It returns a new list, so self.history is untouched until the turn succeeds.
5. send() and stream_events() share validation, request building, and committing. They differ only in how they call the API.
6. stream_events() yields text events while streaming, then exactly one done event, whether the turn succeeded, failed validation, or hit an API error. Interfaces can therefore always rely on receiving a final done event. If the consumer stops iterating early, the generator is closed, the with block closes the stream, and nothing is committed (Lesson 60).
7. _commit() applies the stop-reason rules once: refusals are not stored (Lesson 25), and successful turns are stored as complete content blocks via to_dict() (Lesson 55). new_messages contains exactly the user and assistant messages added by this turn.

Common Mistakes

Putting Chatbot Logic in the Interface

What happens: the CLI, web app, and API each implement history, trimming, and error handling differently, and fixes must be repeated three times.

Fix: keep conversation logic in one core, and keep interfaces thin.

Mutating History Before a Turn Succeeds

What happens: failed or refused turns leave dangling user messages.

Fix: build a pending list, and commit only in one place after success.

Leaking SDK Exceptions to Interfaces

What happens: every interface must know the SDK's exception hierarchy, and unhandled exceptions crash web requests.

Fix: convert errors into a result type, such as Reply(ok=False, error=...).

A Streaming Generator Without a Final Event

What happens: interfaces cannot tell whether the stream finished, failed, or was refused.

Fix: always yield one final event that carries the complete result.

Best Practices

• Separate the core from interfaces.
• Return one result type from every turn.
• Validate input in the core, using short error codes.
• Commit history in one place, and only for usable replies.
• Report the new messages of each turn for storage.
• Offer both blocking and streaming methods that share the same logic.

Exercise

1. Create a second BotConfig for a "SQL helper" bot and run both bots side by side.
2. Add a max_tokens override parameter to send() for single long answers.
3. Write a test that uses a fake client (Lesson 43) to check that a refused reply does not change bot.history.
4. Add a last_reply attribute to ChatBot that stores the most recent Reply.

from types import SimpleNamespace

from chatbot_core import ChatBot


class RefusingClient:
    def __init__(self):
        refusal = SimpleNamespace(
            content=[], stop_reason="refusal",
            usage=SimpleNamespace(input_tokens=10, output_tokens=0),
        )
        self.messages = SimpleNamespace(create=lambda **request: refusal)


bot = ChatBot(RefusingClient())
reply = bot.send("Some request")

assert reply.ok is False and reply.stop_reason == "refusal"
assert bot.history == []
print("Refusals do not change the history.")

The fake client returns a refusal without calling the API. _commit() returns before reading to_dict(), so the SimpleNamespace object does not need that method.