Build your first end-to-end chatbot

It is Wednesday. On Friday at 11 am, Meera will sit in front of six store managers and type questions to Ask Crumb herself. Not slides, not screenshots: a live chat.

Arjun has all the pieces from this module and the last one. He knows the model is stateless, so he must resend history. He has a proper system prompt with Crumb’s facts and rules. He knows how to keep each conversation separate and trim long ones. He knows every response comes with a stop reason and a token receipt. What he does not have is one program that puts it all together.

“Keep it small,” says Priya. “A terminal chatbot. No website, no WhatsApp yet. If the brain works in a terminal, wiring it to WhatsApp later is plumbing. And make it stream: Meera should see the words appear as they are written, not stare at a blank screen for four seconds.”

Meera’s requirements are on a sticky note: Fast. Correct. Can start over. Show me what it costs.

What you are building

By the end of this lab you will have Ask Crumb, a terminal chatbot that:

• answers as Crumb’s assistant, using the system prompt from this module,
• remembers the conversation by resending history,
• streams replies word by word as they are generated,
• shows the tokens used after each reply, plus session totals and estimated cost on request,
• supports /reset to start a fresh conversation, /usage for totals, and /quit to exit,
• trims old messages when the conversation outgrows a token budget,
• runs on Claude, OpenAI or Gemini by changing one line in .env.

flowchart TD
  A["You type a message"] --> B{"Starts with /?"}
  B -->|Yes| C["Run the command: reset, usage or quit"]
  B -->|No| D["Append to history as role user"]
  D --> E["Send system prompt and history, streaming"]
  E --> F["Print each piece of text as it arrives"]
  F --> G["Append the full reply as role assistant"]
  G --> H["Record tokens; trim history if over budget"]
  H --> A
  C --> A

Time: 45 to 60 minutes. Cost: a few rupees of API usage for the whole lab.

Step 1: Set up the project

Create a fresh project, so the demo cannot be broken by anything else on your laptop.

uv init ask-crumb
cd ask-crumb
uv add anthropic openai google-genai python-dotenv
mkdir prompts

Copy your .env from the crumb-ai project in Module 01 (it has all three API keys and the model names), add one new line, and make sure .gitignore lists .env:

# .env  (never commit this file)
OPENAI_API_KEY=sk-...
GEMINI_API_KEY=AIza...
ANTHROPIC_API_KEY=sk-ant-...
OPENAI_MODEL=gpt-5-mini
GEMINI_MODEL=gemini-2.5-flash
PROVIDER=claude

PROVIDER will choose which model family Ask Crumb talks to. You will use it in Step 8.

You should now see an ask-crumb folder with pyproject.toml, .env, a .gitignore that lists .env, and an empty prompts folder. git check-ignore -v .env should print the rule that ignores it.

Step 2: Write the system prompt

Copy the five-section Ask Crumb prompt from the lesson on system prompts (identity, job, facts, rules, style) into prompts/ask_crumb.md. Keep the {today} placeholder in the facts section.

Then create chatbot.py with the code that loads it:

# chatbot.py: Ask Crumb, a terminal chatbot
import datetime
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()

PROMPT = Path("prompts/ask_crumb.md").read_text(encoding="utf-8")

def build_system() -> str:
    today = datetime.date.today().strftime("%A, %d %B %Y")
    return PROMPT.replace("{today}", today)

if __name__ == "__main__":
    print(build_system())

Run uv run chatbot.py.

You should now see the whole system prompt printed, with today’s real date where {today} was. If you see {today} itself, the placeholder in the file is spelled differently from the one in the code.

Step 3: Make one call and read the receipt

Before building a loop, prove a single round trip works, and read the whole response, not only the text. Replace the if __name__ block at the bottom of chatbot.py:

import anthropic

if __name__ == "__main__":
    client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY
    response = client.messages.create(
        model="claude-opus-5-5",
        max_tokens=2000,
        output_config={"effort": "low"},   # chat replies need little thinking: faster and cheaper
        system=build_system(),
        messages=[{"role": "user", "content": "Do you deliver to Lonavala?"}],
    )
    print(next((b.text for b in response.content if b.type == "text"), ""))
    print("stop reason:", response.stop_reason)
    print("tokens:", response.usage.input_tokens, "in,", response.usage.output_tokens, "out")

Why effort: "low"? Claude Opus 5.5 always thinks before answering; the default effort is medium. Customer chat is mostly about facts you supply, not deep reasoning, so low effort gives faster first words and fewer output tokens.

You should now see a short reply saying Crumb delivers only within Pune city limits, a stop reason of end_turn, and roughly 550 to 600 input tokens. Almost all of those input tokens are the system prompt. That number will matter in Step 7.

Step 4: Add the chat loop and the history

Now the heart of every chatbot: a loop that keeps the conversation. Replace the bottom block again:

def main() -> None:
    client = anthropic.Anthropic()
    system = build_system()
    history: list[dict] = []
    print("Ask Crumb is ready. Type /quit to exit.\n")

    while True:
        text = input("You: ").strip()
        if not text:
            continue
        if text == "/quit":
            break

        history.append({"role": "user", "content": text})
        response = client.messages.create(
            model="claude-opus-5-5",
            max_tokens=2000,
            output_config={"effort": "low"},
            system=system,
            messages=history,            # the WHOLE conversation, every time
        )
        reply = next((b.text for b in response.content if b.type == "text"), "")
        history.append({"role": "assistant", "content": reply})
        print(f"Crumb: {reply}")
        print(f"  [{response.usage.input_tokens} in · {response.usage.output_tokens} out]\n")

if __name__ == "__main__":
    main()

Arjun’s first version of this step was missing the history.append({"role": "assistant", ...}) line. The bot answered his first question fine. Then he asked “Why not?” and it replied “Could you tell me what you’re asking about?” It had never seen its own previous answer, because his code never added it. The model only knows what is in the list.

Run it and have a short conversation:

You: Do you have eggless cakes?
Crumb: Yes! Every cake has an eggless version with 24 hours' notice, except our Christmas plum cake, which contains eggs.
  [571 in · 64 out]

You: ok, and for Sunday?
Crumb: Order by Saturday at the same time and we can have an eggless cake ready for Sunday. Which cake would you like?
  [649 in · 58 out]

You should now see the bot answer follow-up questions like “ok, and for Sunday?” correctly, and the input tokens grow on every turn.

Step 5: Stream the replies

Meera wants to see words appearing, not a pause followed by a paragraph. Streaming means the API sends the reply in small pieces as the model generates them, instead of all at once at the end. The total time is about the same; the perceived wait is far shorter, because the first words arrive within a second or so.

Replace the non-streaming call inside the loop with Claude’s streaming helper:

        history.append({"role": "user", "content": text})
        print("Crumb: ", end="", flush=True)
        pieces = []
        with client.messages.stream(
            model="claude-opus-5-5",
            max_tokens=2000,
            output_config={"effort": "low"},
            system=system,
            messages=history,
        ) as stream:
            for chunk in stream.text_stream:          # text arrives in small pieces
                print(chunk, end="", flush=True)      # show each piece immediately
                pieces.append(chunk)
            final = stream.get_final_message()        # the complete response, with usage
        reply = "".join(pieces)
        history.append({"role": "assistant", "content": reply})
        print(f"\n  [{final.usage.input_tokens} in · {final.usage.output_tokens} out]\n")

Line by line where it is new:

• client.messages.stream(...) opens a streaming request. The with block makes sure the connection is closed properly afterwards.
• stream.text_stream yields only the text pieces of the reply, skipping the model’s hidden thinking.
• print(chunk, end="", flush=True) prints each piece without a newline. flush=True forces it onto the screen immediately.
• stream.get_final_message() gives you the complete response object once the stream ends, including stop_reason and usage. Streaming does not cost you the receipt.

Arjun’s first attempt left out flush=True. The reply still appeared all at once at the end, and he spent ten minutes convinced streaming was broken. Priya explained: Python collects printed text in a buffer (a holding area) and writes it to the screen in batches, usually at each newline. Since the pieces had no newlines, nothing appeared until the end. flush=True empties the buffer on every print.

You should now see replies appear word by word, followed by the token line. There may be a short pause before the first word while the model thinks; low effort keeps it brief.

Step 6: Add commands and the usage display

Meera wants to start over between demo questions and see what it all costs. Add three commands and a running total. First, near the top of chatbot.py:

PRICE_IN, PRICE_OUT = 4.00, 20.00   # $ per million tokens, Claude Opus 5.5 at time of writing

def new_state() -> dict:
    return {"history": [], "tokens_in": 0, "tokens_out": 0}

def show_usage(state: dict) -> None:
    cost = (state["tokens_in"] * PRICE_IN + state["tokens_out"] * PRICE_OUT) / 1_000_000
    print(f"  This conversation: {state['tokens_in']:,} in / {state['tokens_out']:,} out, "
          f"about ${cost:.4f} (₹{cost * 88:.2f})\n")

Keeping the history and the counters together in one state dictionary is deliberate. Arjun’s first /reset cleared the history but not the counters, so after a reset /usage still showed the earlier conversation’s cost. With one state, a reset is one line that cannot forget anything: state = new_state().

Then, in the loop, replace the input and command handling:

    state = new_state()
    print("Ask Crumb is ready. Commands: /reset, /usage, /quit\n")

    while True:
        try:
            text = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):   # Ctrl+D or Ctrl+C: exit politely
            print("\nBye!")
            break
        if not text:
            continue
        if text == "/quit":
            break
        if text == "/reset":
            state = new_state()
            print("  (Conversation cleared. Ask Crumb has forgotten everything.)\n")
            continue
        if text == "/usage":
            show_usage(state)
            continue

Use state["history"] instead of history in the rest of the loop, and after each reply add the counts: state["tokens_in"] += final.usage.input_tokens and state["tokens_out"] += final.usage.output_tokens.

The try/except around input() matters on demo day. Without it, pressing Ctrl+C prints a long red error trace. With it, the program says “Bye!” and exits.

You should now see /usage print the totals and an estimated cost (well under a rupee per exchange), /reset make the bot forget your name if you told it, and Ctrl+C exit cleanly.

Step 7: Keep the history inside a budget

A demo chat will be short, but the WhatsApp version will not be. Add the trimming from the multi-turn lesson, with one improvement: instead of estimating tokens from characters, use the exact input count the API just reported.

HISTORY_BUDGET = 3_000   # input tokens per request before we trim
DROP_PAIRS = 3           # trim several pairs at once, not one per turn

def trim(state: dict, last_input_tokens: int) -> None:
    if last_input_tokens <= HISTORY_BUDGET:
        return
    cut = min(DROP_PAIRS * 2, len(state["history"]) - 2)   # whole user+assistant pairs only
    if cut > 0:
        del state["history"][:cut]
        print(f"  (Trimmed {cut} old messages to stay within budget.)\n")

Call trim(state, final.usage.input_tokens) at the end of each turn, after appending the reply.

The * 2 is not decoration. Arjun first wrote cut = DROP_PAIRS, which removed three messages. That left the history starting with an assistant message, and the next request failed with a 400 error saying the first message must be from the user. Removing whole pairs keeps the history alternating and starting with the user.

To test it without a two-hour chat, temporarily set HISTORY_BUDGET = 800 and chat for eight or ten turns.

You should now see a “Trimmed 6 old messages” line once the input passes the budget, and the input count drop on the next turn. Ask about something from the very start of the chat; the bot should no longer know it. That is the trade-off you accepted. (For the WhatsApp version, you would add the running summary and pinned facts from the multi-turn lesson.) Set the budget back to 3,000 when you are done.

Step 8: Swap the engine

Priya’s last request: “Make it run on OpenAI and Gemini too. If Claude has an outage at 10:59 on Friday, I want to change one word and carry on.”

Move the streaming call into its own file, providers.py, with one function per provider. Each takes the same inputs (system prompt and neutral history) and returns the same outputs (reply text, input tokens, output tokens), printing pieces as they arrive.

# providers.py: one streaming function per provider, all with the same shape
import os
import anthropic
from openai import OpenAI
from google import genai
from google.genai import types

def stream_claude(system: str, history: list[dict]) -> tuple[str, int, int]:
    client = anthropic.Anthropic()
    pieces = []
    with client.messages.stream(
        model="claude-opus-5-5", max_tokens=2000, output_config={"effort": "low"},
        system=system, messages=history,
    ) as stream:
        for chunk in stream.text_stream:
            print(chunk, end="", flush=True)
            pieces.append(chunk)
        final = stream.get_final_message()
    return "".join(pieces), final.usage.input_tokens, final.usage.output_tokens

def stream_openai(system: str, history: list[dict]) -> tuple[str, int, int]:
    client = OpenAI()
    stream = client.chat.completions.create(
        # Model names change; check the provider's models page for the current list.
        model=os.getenv("OPENAI_MODEL", "gpt-5-mini"),
        messages=[{"role": "system", "content": system}] + history,
        stream=True,
        stream_options={"include_usage": True},   # ask for token counts at the end
    )
    pieces, usage = [], None
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            piece = chunk.choices[0].delta.content
            print(piece, end="", flush=True)
            pieces.append(piece)
        if chunk.usage:                            # only the final chunk carries usage
            usage = chunk.usage
    return "".join(pieces), usage.prompt_tokens, usage.completion_tokens

def stream_gemini(system: str, history: list[dict]) -> tuple[str, int, int]:
    client = genai.Client()
    contents = [
        {"role": "model" if m["role"] == "assistant" else "user", "parts": [{"text": m["content"]}]}
        for m in history
    ]
    pieces, usage = [], None
    for chunk in client.models.generate_content_stream(
        model=os.getenv("GEMINI_MODEL", "gemini-2.5-flash"),
        contents=contents,
        config=types.GenerateContentConfig(system_instruction=system),
    ):
        if chunk.text:
            print(chunk.text, end="", flush=True)
            pieces.append(chunk.text)
        if chunk.usage_metadata:
            usage = chunk.usage_metadata
    out = (usage.candidates_token_count or 0) + (usage.thoughts_token_count or 0)
    return "".join(pieces), usage.prompt_token_count, out

ADAPTERS = {"claude": stream_claude, "openai": stream_openai, "gemini": stream_gemini}

def stream_reply(system: str, history: list[dict]) -> tuple[str, int, int]:
    return ADAPTERS[os.getenv("PROVIDER", "claude")](system, history)

Library APIs change quickly. If a name has moved, check the docs linked below.

Notice the three differences you met in this module, now side by side: OpenAI wants the system prompt as the first message and needs include_usage to report tokens when streaming; Gemini wants model instead of assistant, text wrapped in parts, and the system prompt in the config; Claude takes system as its own field. Arjun’s first Gemini run failed with a 400 error because he sent assistant roles straight through. The conversion line fixed it.

In chatbot.py, delete the Claude client and the inline streaming code, import from providers import stream_reply, and replace the streaming block with a guarded call:

        state["history"].append({"role": "user", "content": text})
        print("Crumb: ", end="", flush=True)
        try:
            reply, tokens_in, tokens_out = stream_reply(system, state["history"])
        except Exception as err:
            state["history"].pop()   # undo the user message so the history stays valid
            print(f"\n  (Sorry, something went wrong: {type(err).__name__}. Please try again.)\n")
            continue
        state["history"].append({"role": "assistant", "content": reply})
        state["tokens_in"] += tokens_in
        state["tokens_out"] += tokens_out
        print(f"\n  [{tokens_in:,} in · {tokens_out:,} out]\n")
        trim(state, tokens_in)

The pop() on failure is the lesson from the multi-turn lesson: if a request fails after the user message was appended, remove it, or the next attempt sends two user messages in a row.

The cost estimate in /usage uses Claude’s prices. If you run on another provider, put that provider’s prices in PRICE_IN and PRICE_OUT.

Change PROVIDER=openai in .env, run again, and chat. Then try PROVIDER=gemini.

You should now see Ask Crumb answer in all three configurations, streaming each time, with the same commands working. The wording differs between providers, but the facts and rules from the system prompt hold in all three.

# chatbot.py: Ask Crumb, a terminal chatbot
import datetime
from pathlib import Path
from dotenv import load_dotenv
from providers import stream_reply

load_dotenv()

PROMPT = Path("prompts/ask_crumb.md").read_text(encoding="utf-8")
PRICE_IN, PRICE_OUT = 4.00, 20.00   # $ per million tokens, Claude Opus 5.5 at time of writing
HISTORY_BUDGET = 3_000
DROP_PAIRS = 3

def build_system() -> str:
    today = datetime.date.today().strftime("%A, %d %B %Y")
    return PROMPT.replace("{today}", today)

def new_state() -> dict:
    return {"history": [], "tokens_in": 0, "tokens_out": 0}

def show_usage(state: dict) -> None:
    cost = (state["tokens_in"] * PRICE_IN + state["tokens_out"] * PRICE_OUT) / 1_000_000
    print(f"  This conversation: {state['tokens_in']:,} in / {state['tokens_out']:,} out, "
          f"about ${cost:.4f} (₹{cost * 88:.2f})\n")

def trim(state: dict, last_input_tokens: int) -> None:
    if last_input_tokens <= HISTORY_BUDGET:
        return
    cut = min(DROP_PAIRS * 2, len(state["history"]) - 2)
    if cut > 0:
        del state["history"][:cut]
        print(f"  (Trimmed {cut} old messages to stay within budget.)\n")

def main() -> None:
    system = build_system()
    state = new_state()
    print("Ask Crumb is ready. Commands: /reset, /usage, /quit\n")

    while True:
        try:
            text = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nBye!")
            break
        if not text:
            continue
        if text == "/quit":
            break
        if text == "/reset":
            state = new_state()
            print("  (Conversation cleared. Ask Crumb has forgotten everything.)\n")
            continue
        if text == "/usage":
            show_usage(state)
            continue

        state["history"].append({"role": "user", "content": text})
        print("Crumb: ", end="", flush=True)
        try:
            reply, tokens_in, tokens_out = stream_reply(system, state["history"])
        except Exception as err:
            state["history"].pop()
            print(f"\n  (Sorry, something went wrong: {type(err).__name__}. Please try again.)\n")
            continue
        state["history"].append({"role": "assistant", "content": reply})
        state["tokens_in"] += tokens_in
        state["tokens_out"] += tokens_out
        print(f"\n  [{tokens_in:,} in · {tokens_out:,} out]\n")
        trim(state, tokens_in)

if __name__ == "__main__":
    main()

Step 9: Rehearse the demo

On Thursday evening, Priya plays Meera and runs the script they will use on Friday. Run it yourself and check each result:

You should now see every row behave as described, on at least one provider, three times in a row. If one row fails occasionally, that is sampling at work: strengthen the relevant rule in the system prompt and rerun the test bench from the system prompts lesson.

What you built, and what it is not yet

You now have a real chatbot with every essential part: roles, a system prompt with facts and rules, history, streaming, usage tracking, a reset, trimming and provider independence. That is the core of every chat product you have used.

It is not yet a product. It handles one conversation at a time, in one terminal. Its memory disappears when you quit. A network error shows an apology instead of retrying. It cannot look up real orders. Each of those has a home later in the course: web endpoints and storage in Module 05, retries and fallbacks between providers in Module 19, looking up orders with tools in Module 06, and answering from the full handbook with retrieval in Module 09.

Recap

• A chatbot is a loop: read, append as user, send system prompt plus history, stream, append as assistant, record usage, trim.
• Streaming shows words as they are generated; flush=True makes them appear immediately, and the final message still carries usage and the stop reason.
• Keep the history and the counters in one state object, so /reset cannot forget anything.
• Trim using the exact input tokens the API reports, removing whole user and assistant pairs.
• A small adapter per provider hides the differences in roles, system prompt placement and usage fields.
• If a request fails, remove the unanswered user message so the history stays valid.

On Friday, Meera types her first question with six store managers watching. The words appear almost at once and flow across the screen. She asks about eggless cakes, then Sunday delivery, then, deliberately, the peanut question. Ask Crumb refuses to call anything safe and offers to connect her with the team. One of the managers starts clapping before she has finished reading.

Then Meera types /usage. “Eleven questions,” she reads out. “About five rupees.” She looks over at Arjun. “Now. When can it read the handbook, so it stops saying ‘let me check with the team’ about our cancellation policy?”

Priya smiles. “First, we teach him to write much better prompts.”

Go deeper

• Claude docs, streaming messages: https://docs.anthropic.com
• OpenAI docs, streaming responses: https://platform.openai.com/docs
• Gemini API docs, streaming and multi-turn chat: https://ai.google.dev/gemini-api/docs
• Python docs, the print function and flush: https://docs.python.org/3/tutorial/