Building an AI chatbot with Python in 2026 takes less time than most tutorials suggest — and the gap between a working prototype and a production-ready system is three specific additions: persistent memory, a web API layer, and optionally a RAG retrieval pipeline for document-aware responses.
This guide builds all three, in order. You’ll start with a 30-line terminal chatbot using Claude Sonnet 5, add a FastAPI web server that exposes it as an API, and then add a ChromaDB retrieval layer so the chatbot can answer questions about your own documents. Each version is fully working code — not pseudocode, not snippets — that you can run immediately. If you haven’t set up the Claude API yet, start with the Claude API Python Tutorial first. This guide picks up where that one ends.
Setup: Install Everything You Need
pip install anthropic langchain-anthropic langchain-community \
chromadb fastapi uvicorn python-dotenv pypdf# .env
ANTHROPIC_API_KEY=sk-ant-your-key-hereVersion 1: Terminal Chatbot With Memory (10 Minutes)
The simplest working AI chatbot: a Python script that reads input, sends it to Claude with full conversation history, and prints the response. The critical addition over a single-shot API call is the message history loop — the chatbot remembers everything said in the current session.
import os
import anthropic
from dotenv import load_dotenv
load_dotenv()
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
def run_chatbot(
system_prompt: str = "You are a helpful assistant. Be concise.",
model: str = "claude-sonnet-5"
) -> None:
"""
A terminal chatbot with persistent conversation memory.
Type 'quit', 'exit', or press Ctrl+C to stop.
"""
history: list[dict] = []
print(f"Chatbot ready (model: {model}). Type 'quit' to exit.\n")
while True:
try:
user_input = input("You: ").strip()
except (KeyboardInterrupt, EOFError):
print("\nGoodbye!")
break
if user_input.lower() in ("quit", "exit", "q", ""):
print("Goodbye!")
break
# Add user message to conversation history
history.append({"role": "user", "content": user_input})
try:
response = client.messages.create(
model=model,
max_tokens=1024,
system=system_prompt,
messages=history
)
assistant_message = response.content[0].text
# Add Claude's response to history for the next turn
history.append({"role": "assistant", "content": assistant_message})
print(f"\nClaude: {assistant_message}\n")
except anthropic.APIError as e:
print(f"API error: {e}")
# Remove the failed user message from history
history.pop()
if __name__ == "__main__":
run_chatbot(
system_prompt="""You are a Python programming assistant.
Help developers write clean, production-ready Python code.
When showing code, always include error handling."""
)
Run this with python chatbot.py. The chatbot remembers the full conversation — ask a follow-up question and it knows what you were discussing three turns ago. The system_prompt parameter is where you specialize the chatbot for your use case: customer support, coding assistant, document Q&A, or any other role.
Version 2: FastAPI Web Chatbot (15 Minutes)
The terminal chatbot works for local development. A production AI chatbot needs an HTTP API that a frontend, a mobile app, or another service can call. FastAPI is the 2026 standard for Python web APIs: async-native, automatic documentation, and type-safe request/response models.
import os
import uuid
from contextlib import asynccontextmanager
from dotenv import load_dotenv
import anthropic
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
load_dotenv()
# ─── Data models ────────────────────────────────────────────────
class Message(BaseModel):
role: str # "user" or "assistant"
content: str
class ChatRequest(BaseModel):
message: str
session_id: str | None = None # None = new session
stream: bool = False
class ChatResponse(BaseModel):
response: str
session_id: str
input_tokens: int
output_tokens: int
# ─── In-memory session store (use Redis in production) ──────────
sessions: dict[str, list[Message]] = {}
SYSTEM_PROMPT = """You are a helpful AI assistant.
Answer questions clearly and concisely.
If you don't know something, say so."""
# ─── App setup ──────────────────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
print("Chatbot API starting...")
yield
print("Chatbot API shutting down.")
app = FastAPI(
title="AI Chatbot API",
description="Python chatbot powered by Claude Sonnet 5",
version="1.0.0",
lifespan=lifespan
)
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
# ─── Routes ─────────────────────────────────────────────────────
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
"""
Send a message and get a response.
Include session_id to continue an existing conversation.
Omit session_id to start a new one.
"""
# Create new session or retrieve existing one
session_id = request.session_id or str(uuid.uuid4())
if session_id not in sessions:
sessions[session_id] = []
history = sessions[session_id]
# Add the new user message
history.append(Message(role="user", content=request.message))
try:
# Build the messages payload for the API
messages = [{"role": m.role, "content": m.content} for m in history]
response = client.messages.create(
model="claude-sonnet-5",
max_tokens=1024,
system=SYSTEM_PROMPT,
messages=messages
)
assistant_text = response.content[0].text
# Save assistant response to session history
history.append(Message(role="assistant", content=assistant_text))
return ChatResponse(
response=assistant_text,
session_id=session_id,
input_tokens=response.usage.input_tokens,
output_tokens=response.usage.output_tokens
)
except anthropic.APIError as e:
history.pop() # Remove failed user message
raise HTTPException(status_code=502, detail=f"Claude API error: {e}")
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
"""
Stream the chatbot response token by token.
Useful for long responses — the user sees output immediately.
"""
session_id = request.session_id or str(uuid.uuid4())
if session_id not in sessions:
sessions[session_id] = []
history = sessions[session_id]
history.append(Message(role="user", content=request.message))
messages = [{"role": m.role, "content": m.content} for m in history]
async def generate():
full_response = ""
with client.messages.stream(
model="claude-sonnet-5",
max_tokens=1024,
system=SYSTEM_PROMPT,
messages=messages
) as stream:
for text in stream.text_stream:
full_response += text
yield text
# Save completed response to history after streaming finishes
history.append(Message(role="assistant", content=full_response))
return StreamingResponse(generate(), media_type="text/plain")
@app.get("/sessions/{session_id}")
async def get_session(session_id: str) -> dict:
"""Retrieve conversation history for a session."""
if session_id not in sessions:
raise HTTPException(status_code=404, detail="Session not found")
return {
"session_id": session_id,
"message_count": len(sessions[session_id]),
"messages": [m.model_dump() for m in sessions[session_id]]
}
@app.delete("/sessions/{session_id}")
async def delete_session(session_id: str) -> dict:
"""Clear a conversation session."""
sessions.pop(session_id, None)
return {"status": "deleted", "session_id": session_id}
Run and test the API
# Start the server
uvicorn chatbot_api:app --reload --port 8000
# Test with curl
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "Hello! What can you help me with?"}'
# Continue the conversation (use session_id from previous response)
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message": "Tell me more", "session_id": "your-session-id-here"}'
# Interactive docs available at:
# http://localhost:8000/docsThe FastAPI server automatically generates interactive documentation at /docs — paste the URL in your browser and you can test every endpoint without writing any frontend code. The session_id pattern is what makes the chatbot stateful across HTTP requests: each request is stateless at the protocol level, but the session history stored server-side makes the conversation feel continuous.
Version 3: RAG Document Chatbot (Add Your Own Knowledge)
The chatbot above answers questions from Claude’s training data. Add a RAG layer and it answers questions from your documents — product documentation, legal policies, technical manuals, knowledge bases. This is the same pipeline covered in yesterday’s RAG Tutorial, integrated directly into the chatbot endpoint.
# rag_chatbot.py — add this to the FastAPI app above
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
# ─── Build the knowledge base ───────────────────────────────────
def build_knowledge_base(doc_paths: list[str]) -> Chroma:
"""
Index a list of documents into ChromaDB.
Call this once at startup; the index persists to disk.
"""
documents = []
for path in doc_paths:
if path.endswith(".pdf"):
loader = PyPDFLoader(path)
else:
loader = TextLoader(path)
documents.extend(loader.load())
splitter = RecursiveCharacterTextSplitter(
chunk_size=512, chunk_overlap=50
)
chunks = splitter.split_documents(documents)
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chatbot_knowledge"
)
print(f"Indexed {len(chunks)} chunks from {len(doc_paths)} document(s)")
return vectorstore
# ─── RAG-enhanced chat endpoint ─────────────────────────────────
def get_rag_context(vectorstore: Chroma, query: str, k: int = 3) -> str:
"""Retrieve the most relevant document chunks for a query."""
docs = vectorstore.similarity_search(query, k=k)
if not docs:
return ""
return "\n\n---\n\n".join(
f"[Source: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
for doc in docs
)
RAG_SYSTEM_PROMPT = """You are a helpful assistant with access to specific documents.
When answering:
1. Base your response on the provided context when relevant
2. Cite the source document when using retrieved information
3. If the answer isn't in the context, say so and use your general knowledge
Context from documents:
{context}"""
async def chat_with_rag(
user_message: str,
history: list[Message],
vectorstore: Chroma
) -> str:
"""Chat with document context injected into the system prompt."""
# Retrieve relevant context for this specific message
context = get_rag_context(vectorstore, user_message)
# Inject context into system prompt
system = RAG_SYSTEM_PROMPT.format(
context=context if context else "No specific documents retrieved."
)
messages = [{"role": m.role, "content": m.content} for m in history]
response = client.messages.create(
model="claude-sonnet-5",
max_tokens=1024,
system=system,
messages=messages
)
return response.content[0].text
# ─── Initialize at startup ──────────────────────────────────────
# Add your documents here
KNOWLEDGE_BASE_DOCS = [
"your_document.pdf",
"your_knowledge_base.txt",
]
# Build once; comment out and use load_existing() after first run
knowledge_base = build_knowledge_base(KNOWLEDGE_BASE_DOCS)
Production Checklist Before You Deploy
- Replace in-memory sessions with Redis. The
sessionsdictionary in Version 2 resets every time the server restarts. Useredis-pywith a TTL of 24 hours for production: session data persists across restarts and scales across multiple server instances. - Add rate limiting. Without rate limits, a single user can generate hundreds of dollars in API costs. Use
slowapi(FastAPI-compatible) to limit each IP to 10 requests per minute:@limiter.limit("10/minute")on the chat endpoint. - Track costs per session. Log
input_tokensandoutput_tokensfor every API call. A 10-message conversation with 500 tokens per message uses approximately 5,000 total tokens — at Sonnet 5’s $2/$10 pricing, that’s $0.007 per conversation. At 1,000 conversations per day: $7/day. Know your cost per conversation before scaling. - Stream by default for responses longer than one sentence. Users tolerate waiting 2 seconds for a streaming response that starts immediately. They become frustrated waiting 5 seconds for a non-streaming response. Switch the production endpoint to
/chat/streamfor any response that might take more than 2 seconds. - Add the human review gate for consequential actions. If your chatbot can take actions — sending emails, updating records, submitting forms — add an explicit confirmation step before any action executes. The rogue AI agent incidents this week demonstrated exactly what happens when capable models are given external action tools without confirmation gates. Chatbots that only generate text are low-risk. Chatbots with tool access are not.
For the complete Python chatbot ecosystem overview, see UniversoPython’s 2026 Python chatbot tutorial survey.
The Builder’s Takeaway
An AI chatbot with Python in 2026 is a 30-line terminal script that grows into a production FastAPI service with streaming, session management, rate limiting, and optionally a RAG document layer — all on the same code foundation. Version 1 proves the concept in 10 minutes. Version 2 makes it deployable in 15 more. Version 3 makes it genuinely useful for business applications that need answers grounded in specific documents. The five production checklist items are the difference between a demo and a service. The cost tracking is the difference between a service and a sustainable one. Build Version 1 today, Version 2 this week, and Version 3 when you have a specific document corpus to serve.
Continue in This Series
- RAG Tutorial Python 2026 — the complete retrieval pipeline behind Version 3: chunking, embedding, ChromaDB, and quality improvements
- Claude API Python Tutorial — the foundational API patterns this chatbot builds on: system prompts, streaming, cost tracking
- LangChain Tutorial 2026 — the LCEL chain alternative to the raw SDK approach used here
- How to Build an AI Agent With Python — extend the chatbot into a full agent: add tools and autonomous decision-making
- How to Deploy AI Agents — the production deployment checklist once your chatbot is ready to go live
This post is part of The Agentic Protocol’s Work series — the connective infrastructure layer beneath every autonomous pipeline. See also: RAG Tutorial Python 2026.