Use FastAPI with Claude

Why FastAPI

FastAPI is a modern Python web framework built around type hints and async. It is a popular choice for AI applications, because:

This lesson implements the same API as Lesson 67 (designed in Lesson 66) with FastAPI, reusing chat_service.py unchanged. It then shows a fully asynchronous endpoint with AsyncAnthropic.

Setup

python -m pip install fastapi uvicorn itsdangerous anthropic
export APP_SECRET_KEY="$(python -c 'import secrets; print(secrets.token_hex(32))')"
export ANTHROPIC_API_KEY="sk-ant-..."

• uvicorn is the ASGI server that runs FastAPI applications.
• itsdangerous is required by Starlette's SessionMiddleware, which provides the signed session cookie.

Project Structure

fastapi-chatbot/
├── chatbot_core.py, history_store.py, sessions.py, memory.py   # Lessons 61–64
├── chat_service.py                                             # Lesson 66
├── fastapi_app.py                                              # This lesson
└── static/
    └── chat.html                                               # Lesson 66

Sync and Async Endpoints in FastAPI

FastAPI supports two kinds of endpoint functions:

StreamingResponse follows the same rule: a sync generator is iterated in the thread pool, and an async generator runs on the event loop.

Because chat_service.chat_events() is a synchronous generator using the sync client and sqlite3, the chat endpoints are plain def functions. That is correct and efficient: each stream occupies a thread-pool thread while it runs. For very high numbers of simultaneous streams, a fully async stack is more scalable (see the second example).

Practical Example: fastapi_app.py

"""fastapi_app.py: the Lesson 66 web chatbot implemented with FastAPI."""

import os
import uuid
from contextlib import asynccontextmanager

import anthropic
from anthropic import AsyncAnthropic
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from pydantic import BaseModel, Field
from starlette.middleware.sessions import SessionMiddleware

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


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.async_client = AsyncAnthropic()      # Used by the async endpoint below
    yield
    await app.state.async_client.close()


app = FastAPI(lifespan=lifespan)
app.add_middleware(SessionMiddleware, secret_key=os.environ["APP_SECRET_KEY"], same_site="lax")


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=10_000)


def current_user(request: Request) -> str:
    if "user_id" not in request.session:
        request.session["user_id"] = uuid.uuid4().hex
    return request.session["user_id"]


def error(code: str, message: str, status: int) -> JSONResponse:
    return JSONResponse({"error": {"code": code, "message": message}}, status_code=status)


STREAM_HEADERS = {"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}


@app.get("/")
def index():
    return FileResponse("static/chat.html")


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


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


@app.get("/api/sessions/{session_id}/messages")
def get_messages(session_id: str, request: Request):
    try:
        return {"messages": session_messages(current_user(request), 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, body: ChatRequest, request: Request):
    if not body.message.strip():
        return error("empty_message", "Please type a message.", 400)
    user_id = current_user(request)
    try:
        check_session(user_id, session_id)
    except SessionNotFound:
        return error("session_not_found", "That conversation was not found.", 404)
    return StreamingResponse(chat_events(user_id, session_id, body.message),
                             media_type="text/event-stream", headers=STREAM_HEADERS)


# ---------- A fully asynchronous endpoint (stateless) ----------

class AskRequest(BaseModel):
    question: str = Field(min_length=1, max_length=2_000)


@app.post("/api/ask")
async def ask(body: AskRequest, request: Request):
    client: AsyncAnthropic = request.app.state.async_client

    async def events():
        try:
            async with client.messages.stream(
                model="claude-opus-5-5",
                max_tokens=1024,
                system="You are a concise programming tutor. Answer in under 100 words.",
                messages=[{"role": "user", "content": body.question}],
            ) as stream:
                async for text in stream.text_stream:
                    if await request.is_disconnected():
                        return
                    yield sse({"type": "text", "text": text})
                final = await stream.get_final_message()
            yield sse({"type": "done", "truncated": final.stop_reason == "max_tokens"})
        except anthropic.APIError:
            yield sse({"type": "error", "message": "Something went wrong. Please try again."})

    return StreamingResponse(events(), media_type="text/event-stream", headers=STREAM_HEADERS)

Run the server and open http://127.0.0.1:8000. Interactive API documentation is available at http://127.0.0.1:8000/docs.

uvicorn fastapi_app:app --reload

How It Works

1. lifespan runs once at startup and once at shutdown. It creates one shared AsyncAnthropic client for the async endpoint, and closes it cleanly when the server stops (Lesson 79).
2. SessionMiddleware stores the session in a signed cookie, using APP_SECRET_KEY. same_site="lax" limits cross-site requests, as in the Flask version.
3. ChatRequest is a Pydantic model. FastAPI parses the JSON body into it and validates it automatically: a missing or non-string message, an empty string, or more than 10,000 characters returns a 422 error before the endpoint runs. The extra strip() check catches messages that contain only whitespace.
4. current_user() gives each browser a random ID stored in the signed session, as in Lesson 67.
5. The session endpoints are def functions, so FastAPI runs them in its thread pool, where the synchronous sqlite3 calls in chat_service.py are safe.
6. post_chat() checks ownership before streaming, then returns StreamingResponse with the synchronous chat_events() generator. FastAPI iterates it in the thread pool, so the event loop is never blocked. The short-lived database connections in chat_service.py (Lesson 66) make this safe even when successive chunks run on different threads.
7. /api/ask shows the fully asynchronous alternative for a stateless question. async with and async for stream without blocking (Lesson 56), request.is_disconnected() stops the stream when the browser leaves (Lesson 60), and errors are reported as an error event.

Note: FastAPI's automatic validation errors use FastAPI's own format, {"detail": [...]}, rather than this API's {"error": {...}} shape. Lesson 69 adds an exception handler that converts them, so the browser always receives one consistent error format.

Testing with FastAPI's Test Client

FastAPI's TestClient calls your application directly, without starting a server. Combined with a fake client (Lesson 43), it tests endpoints without API calls:

from fastapi.testclient import TestClient

from fastapi_app import app

with TestClient(app) as client:
    created = client.post("/api/sessions")
    assert created.status_code == 201

    missing = client.get("/api/sessions/does-not-exist/messages")
    assert missing.status_code == 404
    assert missing.json()["error"]["code"] == "session_not_found"

    invalid = client.post(f"/api/sessions/{created.json()['id']}/chat", json={"message": ""})
    assert invalid.status_code == 422
print("API tests passed.")

TestClient keeps cookies between requests, so the session created in the first call belongs to the same test user in later calls. Using with TestClient(app) also runs the lifespan startup and shutdown.

Flask or FastAPI?

Both implement the same API with the same chat service, because the chat logic does not depend on the framework.

Common Mistakes

Blocking Calls in async def Endpoints

What happens: a synchronous Claude call or database query inside async def freezes the event loop, so all other requests wait.

Fix: use def for synchronous work, or use async libraries (AsyncAnthropic) in async def endpoints.

Creating a Client per Request

What happens: extra latency and resource use.

Fix: create shared clients once, at module level or in lifespan.

Forgetting itsdangerous

What happens: SessionMiddleware fails at import or startup.

Fix: install itsdangerous.

Returning Errors Inside the Stream for Invalid Requests

What happens: the browser receives status 200 for invalid input.

Fix: validate with Pydantic and check ownership before returning StreamingResponse.

Mixed Error Formats

What happens: the frontend handles validation errors (detail) and application errors (error) differently.

Fix: convert validation errors to your error format (Lesson 69).

Best Practices

• Use Pydantic models for request bodies, with length limits.
• Match endpoint style to the code inside: def for synchronous code, async def for async code.
• Manage shared clients in lifespan, and close them on shutdown.
• Validate and authorize before streaming.
• Detect disconnected clients in long async streams.
• Test endpoints with TestClient, without real API calls where possible.
• Run behind a production setup, for example Uvicorn workers behind a reverse proxy with HTTPS (Lesson 189).

Exercise

1. Open http://127.0.0.1:8000/docs and send a chat request from the interactive documentation. How is the streamed response displayed?
2. Add DELETE /api/sessions/{session_id} returning status 204.
3. Rewrite get_sessions as a dependency: create a get_user_id(request: Request) -> str function and use user_id: str = Depends(get_user_id) in the endpoints.
4. Explain why post_chat() is a def function, while ask() is an async def function.

Step 3:

from fastapi import Depends


def get_user_id(request: Request) -> str:
    return current_user(request)


@app.get("/api/sessions")
def get_sessions(user_id: str = Depends(get_user_id)):
    return {"sessions": list_sessions(user_id)}

Depends() tells FastAPI to call get_user_id() for each request and pass the result into the endpoint. Endpoints no longer need the request parameter only to find the user, and the identity logic lives in one place.

Step 4: post_chat() uses chat_events(), which calls the synchronous Anthropic client and sqlite3. As a def endpoint, it runs in the thread pool, where blocking calls are safe. ask() uses only async code (AsyncAnthropic, async with, async for, await), so it runs on the event loop and handles many concurrent streams efficiently. Using async def for blocking code would freeze the server, and using def for async-only code would be impossible, because await requires async def.