Use Flask with Claude

Why Flask

Flask is a lightweight Python web framework. It gives you routing, request and response handling, sessions, and templates, and leaves everything else to you. That simplicity makes it a good fit for small chat applications, internal tools, and learning how web chatbots work.

This lesson implements the web chatbot API designed in Lesson 66 with Flask, using the synchronous Anthropic client through the shared chat_service.py module.

Setup

python -m pip install flask anthropic

Flask signs its session cookie with a secret key. Generate one and keep it in an environment variable, never in code (Lesson 8):

python -c "import secrets; print(secrets.token_hex(32))"
export FLASK_SECRET_KEY="paste-the-generated-value-here"
export ANTHROPIC_API_KEY="sk-ant-..."

Project Structure

flask-chatbot/
├── chatbot_core.py      # Lesson 61
├── history_store.py     # Lesson 62
├── sessions.py          # Lesson 63
├── memory.py            # Lesson 64
├── chat_service.py      # Lesson 66
├── flask_app.py         # This lesson
└── static/
    └── chat.html        # Lesson 66

Streaming Responses in Flask

Flask streams a response when you return a Response built from a generator: each yield sends one piece of the body to the browser.

from flask import Response, stream_with_context

return Response(stream_with_context(generator()), mimetype="text/event-stream")

• mimetype="text/event-stream" tells the browser the body is a stream of server-sent events.
• stream_with_context() keeps Flask's request context, including session and request, available while the generator runs. Without it, accessing them inside the generator raises an error, because the request context normally ends when the view function returns.

Response headers help streaming work behind proxies:

Practical Example: flask_app.py

"""flask_app.py: the Lesson 66 web chatbot implemented with Flask."""

import os
import uuid

from flask import Flask, Response, jsonify, request, send_from_directory, session, stream_with_context

from chat_service import (check_session, chat_events, create_session,
                          list_sessions, session_messages)
from sessions import SessionNotFound

MAX_MESSAGE_CHARS = 10_000

app = Flask(__name__, static_folder="static")
app.config.update(
    SECRET_KEY=os.environ["FLASK_SECRET_KEY"],
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SAMESITE="Lax",
)


def current_user() -> str:
    """Return this browser's user ID, creating one on the first visit."""
    if "user_id" not in session:
        session["user_id"] = uuid.uuid4().hex
    return session["user_id"]


def error(code: str, message: str, status: int):
    return jsonify({"error": {"code": code, "message": message}}), status


@app.get("/")
def index():
    return send_from_directory(app.static_folder, "chat.html")


@app.get("/api/sessions")
def get_sessions():
    return jsonify({"sessions": list_sessions(current_user())})


@app.post("/api/sessions")
def post_session():
    return jsonify({"id": create_session(current_user())}), 201


@app.get("/api/sessions/<session_id>/messages")
def get_messages(session_id: str):
    try:
        return jsonify({"messages": session_messages(current_user(), session_id)})
    except SessionNotFound:
        return error("session_not_found", "That conversation was not found.", 404)


@app.post("/api/sessions/<session_id>/chat")
def post_chat(session_id: str):
    data = request.get_json(silent=True) or {}
    text = data.get("message")
    if not isinstance(text, str) or not text.strip():
        return error("empty_message", "Please type a message.", 400)
    if len(text) > MAX_MESSAGE_CHARS:
        return error("message_too_long", "That message is too long. Please shorten it.", 400)

    user_id = current_user()
    try:
        check_session(user_id, session_id)
    except SessionNotFound:
        return error("session_not_found", "That conversation was not found.", 404)

    return Response(
        stream_with_context(chat_events(user_id, session_id, text)),
        mimetype="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


if __name__ == "__main__":
    app.run(debug=True, threaded=True)

Run it and open http://127.0.0.1:5000:

python flask_app.py

How It Works

1. Configuration: SECRET_KEY comes from the environment and signs the session cookie, so users cannot alter their user ID. SESSION_COOKIE_HTTPONLY hides the cookie from JavaScript, and SESSION_COOKIE_SAMESITE="Lax" stops browsers from sending it with cross-site POST requests, which protects against CSRF.
2. current_user() gives each browser a random user ID on its first visit and stores it in the signed session. A real application would use its login system instead (Lesson 66).
3. error() returns every error in the same JSON shape with an appropriate status code, which the page's api() helper displays (Lessons 69 and 70).
4. The read endpoints (GET /api/sessions, GET .../messages) call the chat service and return JSON. SessionNotFound becomes a 404 with the same message whether the session is missing or belongs to someone else (Lesson 63).
5. post_chat() validates the request before streaming:
   • request.get_json(silent=True) returns None instead of raising an error for invalid JSON, and or {} turns that into an empty dictionary;
   • the message must be a non-empty string within the length limit;
   • check_session() verifies ownership. Only then does it return the streaming response. Errors found before streaming get proper status codes.
6. The stream comes from chat_events() (Lesson 66), which sends text, done, or error events and saves the turn. The view itself contains no chat logic.
7. threaded=True lets Flask's development server handle several requests at once, so one user's long stream does not block everyone else.

Note: app.run(debug=True) is for development only. In production, run Flask under a WSGI server such as Gunicorn (gunicorn -w 4 --threads 8 flask_app:app), behind a reverse proxy such as nginx, with HTTPS. With multiple threads per worker, each streaming response occupies a thread while it runs, so size the thread count for the number of simultaneous chats you expect.

Testing the API Without a Browser

curl can exercise each endpoint. -c and -b store and send the session cookie, and -N disables output buffering so the stream appears live:

curl -s -c cookies.txt -b cookies.txt -X POST http://127.0.0.1:5000/api/sessions
# {"id":"3f9c2a7e41d84b6f9c1e0a5b7d2f8c61"}

curl -N -s -c cookies.txt -b cookies.txt \
  -H "Content-Type: application/json" \
  -d '{"message": "What is a Python decorator?"}' \
  http://127.0.0.1:5000/api/sessions/3f9c2a7e41d84b6f9c1e0a5b7d2f8c61/chat

Example stream:

data: {"type": "text", "text": "A decorator is a function"}

data: {"type": "text", "text": " that wraps another function..."}

data: {"type": "done", "truncated": false}

Flask Within a Larger Project

Common Mistakes

Hardcoding the Secret Key

What happens: anyone who sees the code can forge session cookies.

Fix: load SECRET_KEY from the environment, and keep it secret.

Accessing session or request in a Generator Without stream_with_context

What happens: RuntimeError: Working outside of request context.

Fix: wrap the generator in stream_with_context(), or read everything you need before returning the response, as post_chat() does.

Validating Inside the Stream

What happens: invalid input can only be reported as a stream event with status 200.

Fix: validate and check ownership before returning the streaming Response.

Running the Development Server in Production

What happens: poor performance, no hardening, and debug mode can expose an interactive debugger.

Fix: use a production WSGI server, and never enable debug=True in production.

Buffering Proxies

What happens: the answer arrives all at once instead of streaming.

Fix: send X-Accel-Buffering: no and Cache-Control: no-cache, and check your proxy's buffering settings.

Best Practices

• Keep views thin: validate, identify the user, call the chat service, and return.
• Load secrets from the environment, and use secure cookie settings.
• Validate before streaming, with clear JSON errors.
• Stream with Response and stream_with_context, with anti-buffering headers.
• Use a production WSGI server with enough threads for concurrent streams.
• Add rate limiting and real authentication before going public.

Exercise

1. Add DELETE /api/sessions/<session_id> using the delete_session() function from Lesson 66's exercise, returning 204 on success.
2. Move the chat routes into a Blueprint named chat, registered with the prefix /api.
3. Add a simple per-user limit of 20 chat requests per hour, stored in memory, that returns 429 with the error code rate_limited.
4. Test the API with curl, including an empty message and a session ID that does not exist.

Step 1:

from chat_service import delete_session


@app.delete("/api/sessions/<session_id>")
def remove_session(session_id: str):
    try:
        delete_session(current_user(), session_id)
    except SessionNotFound:
        return error("session_not_found", "That conversation was not found.", 404)
    return "", 204

Step 3:

import time
from collections import defaultdict, deque

recent_requests: dict[str, deque] = defaultdict(deque)


def rate_limited(user_id: str, limit: int = 20, window: int = 3600) -> bool:
    now = time.time()
    timestamps = recent_requests[user_id]
    while timestamps and timestamps[0] < now - window:
        timestamps.popleft()
    if len(timestamps) >= limit:
        return True
    timestamps.append(now)
    return False

At the start of post_chat(), after user_id = current_user():

if rate_limited(user_id):
    return error("rate_limited", "You're sending messages too quickly. Please wait a while.", 429)

A deque holds the request times; old entries are removed from the left. This in-memory version works for a single process. With several workers or servers, store the counts in a shared store such as Redis.