← Appendix A: Python 🏠 Home Appendix C: PySpark →

Appendix B: FastAPI Essentials for Agent Deployment

Starting in M08 and peaking in M21–M22B, you'll deploy your agents as production APIs. This appendix covers the FastAPI skills you'll need — from your first endpoint to Docker deployment and SSE streaming.

Where FastAPI Appears in This Course

M08 (Conversation Management) — streaming agent responses via SSE. M21 (API Design & Deployment) — full REST API with Cloud Run and Lambda deployment. M22B (Deploy Lab) — Dockerized agent APIs. Capstones 5A/5B/5C/6 — production agent systems with WebSockets, background tasks, and multi-container orchestration.

What You'll Learn

  • Build a FastAPI application with GET and POST routes
  • Use Pydantic models for automatic request/response validation
  • Handle errors with HTTPException and custom handlers
  • Stream agent responses to clients with Server-Sent Events (SSE)
  • Add WebSocket support for real-time bidirectional communication
  • Deploy with uvicorn and Docker

What Is FastAPI?

Everyday Analogy

Before: Imagine you've built a brilliant cooking robot that can prepare any dish — but it's sitting in your garage. Nobody can order food from it. Pain: Your AI agent runs great locally, but it's trapped on your laptop. Other services, frontend apps, and teammates can't use it. Mapping: FastAPIA modern, high-performance Python web framework for building APIs. It uses type hints for automatic validation, generates interactive documentation, and supports async natively. is like building a restaurant around your robot — it adds a front door (HTTP endpoints), a menu (API docs), and a kitchen ticket system (request validation) so anyone can order from your agent over the internet.

Technical Definition

FastAPI is a Python web framework built on StarletteAn async web framework that provides the HTTP layer (routing, middleware, WebSockets) that FastAPI builds on top of. (async HTTP) and PydanticA data validation library that uses Python type hints to automatically check that incoming data matches expected types and constraints. (data validation). It automatically generates OpenAPI documentation, validates all incoming data, and supports async/await natively — making it the natural choice for wrapping async Claude API calls in a production endpoint.

FastAPI Request Lifecycle
🌐Client Request — POST /chat with JSON body
↓
🛡️Middleware — CORS headers, logging, auth checks
↓
📋Pydantic Validation — auto-validate request body against model
↓
⚙️Route Handler — your async function runs (calls Claude API)
↓
📦Response Serialization — Pydantic model → JSON response
↓
✅Client Response — 200 OK with validated JSON

Setup & First App

Terminal 
# Install FastAPI and uvicorn (the ASGI server)
pip install fastapi uvicorn[standard]
Python — server.py 
from fastapi import FastAPI

# WHAT: Create a FastAPI application instance
# WHY: This is the central object — all routes attach to it
app = FastAPI(
    title="My Agent API",
    description="A Claude-powered agent",
    version="1.0.0",
)

# WHAT: Define a GET endpoint at /health
# WHY: Health checks are required by every cloud platform (Cloud Run, ECS, K8s)
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

# WHAT: Define a POST endpoint at /chat
# WHY: POST carries a request body — GET doesn't
@app.post("/chat")
async def chat(message: str):
    return {"response": f"You said: {message}"}
Terminal 
# Run the server
uvicorn server:app --reload --port 8000

# --reload   → auto-restart on code changes (dev only)
# server:app → file "server.py", object "app"

# Open http://localhost:8000/docs for interactive API documentation
What Just Happened?

UvicornAn ASGI (Asynchronous Server Gateway Interface) server that runs your FastAPI app. It handles incoming HTTP connections and routes them to your async Python functions. starts listening on port 8000. When a request hits /health, FastAPI calls your health_check() function and serializes the returned dict to JSON automatically. The /docs endpoint gives you free interactive Swagger documentation — you can test every endpoint right in the browser.

Routes & HTTP Methods

Python 
from fastapi import FastAPI

app = FastAPI()

# GET — retrieve data (no body)
@app.get("/agents")
async def list_agents():
    return [{"id": "agent-1", "name": "Research Bot"}]

# POST — create or process (with body)
@app.post("/agents")
async def create_agent(name: str):
    return {"id": "agent-2", "name": name}

# PUT — full update
@app.put("/agents/{agent_id}")
async def update_agent(agent_id: str, name: str):
    return {"id": agent_id, "name": name}

# DELETE — remove
@app.delete("/agents/{agent_id}")
async def delete_agent(agent_id: str):
    return {"deleted": agent_id}

# WHAT: {agent_id} is a path parameter — FastAPI extracts it automatically
# WHY: RESTful APIs use path params to identify resources

# The pattern you'll use most: POST /chat with a JSON body
# That's the "front door" to your agent

Request & Response Models

This is where Pydantic (Appendix A) meets FastAPI. You define models, and FastAPI validates every incoming request automatically — no manual if checks.

Python 
from fastapi import FastAPI
from pydantic import BaseModel, Field
import anthropic

app = FastAPI()
client = anthropic.Anthropic()

# WHAT: Define the shape of incoming requests
# WHY: FastAPI auto-rejects malformed requests with clear 422 errors
class ChatRequest(BaseModel):
    message: str = Field(..., min_length=1, max_length=10000)
    conversation_id: str | None = None
    model: str = "claude-sonnet-4-6"
    max_tokens: int = Field(default=1024, ge=1, le=4096)

# WHAT: Define the shape of outgoing responses
# WHY: Documents your API and ensures you don't leak internal fields
class ChatResponse(BaseModel):
    response: str
    conversation_id: str
    tokens_used: int
    model: str

# WHAT: Use the model as the type hint for the body parameter
# WHY: FastAPI reads the type hint, validates the body, and injects the parsed object
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
    """Send a message to the agent."""
    response = client.messages.create(
        model=request.model,
        max_tokens=request.max_tokens,
        messages=[{"role": "user", "content": request.message}],
    )

    return ChatResponse(
        response=response.content[0].text,
        conversation_id=request.conversation_id or "new-convo",
        tokens_used=response.usage.input_tokens + response.usage.output_tokens,
        model=response.model,
    )

# GOTCHA: If a client sends {"message": "", "max_tokens": 99999},
# FastAPI returns 422 with details about BOTH validation failures
# before your handler code even runs.
Why It Matters for Agents

In M21, you'll deploy an agent API that external clients call. Without Pydantic models, you'd write dozens of lines of validation code. With them, you declare the rules once and FastAPI enforces them for every request — including generating interactive docs at /docs that show clients exactly what to send. It's validation, documentation, and serialization in one class.

Path & Query Parameters

Python 
from fastapi import FastAPI, Query

app = FastAPI()

# Path parameters — part of the URL
# GET /conversations/abc-123
@app.get("/conversations/{conversation_id}")
async def get_conversation(conversation_id: str):
    return {"id": conversation_id, "messages": []}

# Query parameters — after the ? in the URL
# GET /conversations?limit=10&offset=0
@app.get("/conversations")
async def list_conversations(
    limit: int = Query(default=10, ge=1, le=100),
    offset: int = Query(default=0, ge=0),
    search: str | None = Query(default=None, max_length=200),
):
    return {
        "limit": limit,
        "offset": offset,
        "search": search,
        "results": [],
    }

# Combine both
# GET /agents/agent-1/logs?since=2026-01-01
@app.get("/agents/{agent_id}/logs")
async def get_agent_logs(agent_id: str, since: str | None = None):
    return {"agent_id": agent_id, "since": since, "logs": []}

Error Handling

Python 
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import anthropic

app = FastAPI()
client = anthropic.Anthropic()

# WHAT: HTTPException returns a proper error response with status code
# WHY: Clients need structured errors, not Python tracebacks

@app.post("/chat")
async def chat(message: str):
    if not message.strip():
        # WHAT: 400 = bad request (client's fault)
        raise HTTPException(status_code=400, detail="Message cannot be empty")

    try:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": message}],
        )
        return {"response": response.content[0].text}

    except anthropic.RateLimitError:
        # WHAT: 429 = too many requests
        raise HTTPException(
            status_code=429,
            detail="Rate limited. Please retry after a few seconds.",
            headers={"Retry-After": "5"},
        )

    except anthropic.AuthenticationError:
        # WHAT: 500 = server error (our config is wrong, not client's fault)
        raise HTTPException(status_code=500, detail="Agent misconfigured")

    except anthropic.APIError as e:
        raise HTTPException(status_code=502, detail=f"Upstream API error: {e}")


# WHAT: Global exception handler — catches anything you missed
# WHY: Never leak Python tracebacks to clients
@app.exception_handler(Exception)
async def global_handler(request, exc):
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"},
    )
Common Status Codes for Agent APIs

200 — Success. 400 — Bad request (invalid input). 401 — Unauthorized (missing/bad API key). 404 — Conversation or agent not found. 422 — Validation error (Pydantic auto-generates this). 429 — Rate limited. 500 — Server error. 502 — Upstream (Claude API) error.

SSE Streaming

Everyday Analogy

Before: Imagine ordering food and the waiter disappears for 30 minutes, then brings everything at once. Pain: Without streaming, your user stares at a blank screen while Claude generates a 2,000-token response — latency that feels like the app is frozen. Mapping: Server-Sent Events (SSE)A web standard where the server sends a continuous stream of text events to the client over a single HTTP connection. Unlike WebSockets, SSE is one-directional (server to client) and works over plain HTTP. are like a waiter who brings each course as it's ready — the user sees tokens appear in real time, exactly like ChatGPT or Claude.ai.

Python 
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import anthropic
import json

app = FastAPI()
client = anthropic.Anthropic()

class ChatRequest(BaseModel):
    message: str

# WHAT: A generator that yields SSE-formatted chunks
# WHY: StreamingResponse calls this repeatedly, sending each chunk to the client
async def stream_agent_response(message: str):
    with client.messages.stream(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": message}],
    ) as stream:
        for text in stream.text_stream:
            # SSE format: "data: ...\n\n"
            yield f"data: {json.dumps({'text': text})}\n\n"

    # Signal the stream is done
    yield f"data: {json.dumps({'done': True})}\n\n"


@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    return StreamingResponse(
        stream_agent_response(request.message),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

# Client-side (JavaScript):
# const source = new EventSource('/chat/stream');
# source.onmessage = (e) => {
#   const data = JSON.parse(e.data);
#   if (data.done) { source.close(); return; }
#   document.getElementById('output').textContent += data.text;
# };

WebSockets

SSE is one-directional (server → client). WebSocketsA protocol that provides full-duplex (bidirectional) communication over a single TCP connection. Both client and server can send messages at any time without polling. provide full-duplex communication — used in Capstone 5B for real-time agent interactions.

Python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import anthropic
import json

app = FastAPI()
client = anthropic.Anthropic()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    # WHAT: Accept the WebSocket connection
    await websocket.accept()
    messages = []  # conversation history

    try:
        while True:
            # WHAT: Wait for a message from the client
            data = await websocket.receive_text()
            user_msg = json.loads(data)

            messages.append({"role": "user", "content": user_msg["message"]})

            # Call Claude and stream the response back
            with client.messages.stream(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                messages=messages,
            ) as stream:
                full_response = ""
                for text in stream.text_stream:
                    full_response += text
                    await websocket.send_json({"type": "chunk", "text": text})

            messages.append({"role": "assistant", "content": full_response})
            await websocket.send_json({"type": "done"})

    except WebSocketDisconnect:
        print("Client disconnected")

Background Tasks

Sometimes you need to do work after sending the response — like logging, analytics, or kicking off a long-running agent workflow.

Python 
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import json

app = FastAPI()

class ChatRequest(BaseModel):
    message: str
    conversation_id: str

# WHAT: A regular function that runs after the response is sent
# WHY: Don't make the client wait for logging or analytics
def log_interaction(conversation_id: str, message: str, response: str):
    log_entry = {
        "conversation_id": conversation_id,
        "message": message,
        "response": response[:200],
    }
    with open("interactions.jsonl", "a") as f:
        f.write(json.dumps(log_entry) + "\n")


@app.post("/chat")
async def chat(request: ChatRequest, background_tasks: BackgroundTasks):
    # Process the request normally
    response_text = "Agent response here..."

    # WHAT: Schedule work to happen AFTER the response is sent
    # WHY: The client gets a fast response; logging happens asynchronously
    background_tasks.add_task(
        log_interaction,
        request.conversation_id,
        request.message,
        response_text,
    )

    return {"response": response_text}  # sent immediately

Middleware & CORS

Python 
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# WHAT: CORS middleware lets browser frontends call your API
# WHY: Without this, a React app on localhost:3000 can't call your API on :8000
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://myapp.com"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

# WHAT: Custom middleware — runs on EVERY request
# WHY: Add logging, timing, auth checks across all endpoints
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{elapsed:.3f}s"
    return response
Security Warning

Never use allow_origins=["*"] in production. This allows any website to call your agent API, which could lead to abuse. Restrict to your actual frontend domains.

Docker Deployment

M22B walks through full Docker deployment. Here's the pattern you'll use:

Dockerfile 
FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# WHAT: Run with multiple workers for production
# WHY: One worker handles one request at a time — 4 workers = 4 concurrent requests
# GOTCHA: --host 0.0.0.0 is required inside Docker (not 127.0.0.1)
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
Terminal 
# Build and run
docker build -t my-agent-api .
docker run -p 8000:8000 -e ANTHROPIC_API_KEY=$ANTHROPIC_API_KEY my-agent-api

# Test it
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!"}'
Key Gotcha

Pass your ANTHROPIC_API_KEY via the -e flag or Docker Compose environment: section. Never bake API keys into the Docker image — they'd be visible to anyone who pulls it.

Testing Your API

Python — test_server.py 
from fastapi.testclient import TestClient
from server import app

# WHAT: TestClient simulates HTTP requests without starting a real server
# WHY: Fast, deterministic tests — no network, no ports, no Docker

client = TestClient(app)

def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "healthy"}

def test_chat():
    response = client.post("/chat", json={"message": "Hello!"})
    assert response.status_code == 200
    assert "response" in response.json()

def test_validation_rejects_empty():
    response = client.post("/chat", json={"message": ""})
    assert response.status_code == 422  # Pydantic validation error

def test_validation_rejects_long():
    response = client.post("/chat", json={"message": "x" * 20000})
    assert response.status_code == 422
Terminal 
# Run tests
pip install pytest
pytest test_server.py -v

Knowledge Check

1. What does uvicorn server:app --reload mean?

ARun the app function in server.py as a background daemon
BStart the ASGI server using the app object from server.py, auto-restarting on file changes
CReload the server module and run all its tests
DCompile server.py to a binary called app
Correct! server:app tells uvicorn to import app from server.py. The --reload flag watches for file changes and restarts automatically (dev only).
The syntax is module:variable — uvicorn imports the app object from server.py and serves it. --reload enables auto-restart on file changes for development.

2. What HTTP status code does FastAPI return when Pydantic validation fails?

A400 Bad Request
B422 Unprocessable Entity
C500 Internal Server Error
D415 Unsupported Media Type
Correct! FastAPI returns 422 with detailed error messages showing exactly which fields failed validation and why.
FastAPI returns 422 Unprocessable Entity for Pydantic validation errors. This is distinct from 400 (generic bad request) and includes structured details about each validation failure.

3. What's the difference between SSE (StreamingResponse) and WebSockets?

ASSE is faster; WebSockets add overhead
BWebSockets only work in browsers; SSE works everywhere
CSSE is one-directional (server to client); WebSockets are bidirectional
DSSE supports binary data; WebSockets are text-only
Correct! SSE streams data from server to client over plain HTTP. WebSockets allow both sides to send messages at any time. For agent "streaming responses," SSE is usually sufficient. WebSockets are needed when the client must send multiple messages during a stream.
SSE is one-directional (server → client) and uses plain HTTP. WebSockets are bidirectional — both sides can send at any time. For streaming Claude responses, SSE is usually enough.

4. Why is --host 0.0.0.0 required when running uvicorn inside Docker?

AIt enables HTTPS encryption
BIt disables authentication for testing
CIt binds to all network interfaces, allowing connections from outside the container
DIt increases the maximum number of workers
Correct! By default, uvicorn binds to 127.0.0.1 (localhost only). Inside Docker, the host machine connects through the container's network interface, so you must bind to 0.0.0.0 (all interfaces) for the port mapping to work.
Inside Docker, 127.0.0.1 means "only inside this container." To accept connections from outside (via docker run -p 8000:8000), you must bind to 0.0.0.0 which means "all network interfaces."

5. What does BackgroundTasks do in FastAPI?

ARuns tasks before processing the request
BRuns tasks after the response is sent, without blocking the client
CCreates a separate thread pool for CPU-heavy work
DSchedules tasks to run on a cron schedule
Correct! BackgroundTasks queues work that runs after the HTTP response is already sent. The client gets a fast response while logging, analytics, or cleanup happen asynchronously.
BackgroundTasks runs functions after the response is sent. The client gets the response immediately; the background work (logging, analytics) happens without blocking.

Summary

Key Takeaways

  • FastAPI + Pydantic — declare request/response models and get automatic validation, serialization, and docs.
  • Async native — all route handlers can be async def, perfect for wrapping async Claude API calls.
  • SSE streaming — StreamingResponse with a generator yields tokens in real time, like ChatGPT.
  • WebSockets — full-duplex communication for multi-turn agent conversations.
  • Error handling — HTTPException for structured errors; global handlers as a safety net.
  • Docker deployment — --host 0.0.0.0, multiple workers, env vars for secrets.
  • TestClient — test your API without starting a real server.

Where to Go Next

You'll first use FastAPI in M08 (streaming conversations), then go deep in M21 (full API deployment to Cloud Run and Lambda) and M22B (Dockerized deployment lab). The capstones add WebSockets, background tasks, and multi-container orchestration.

← Appendix A: Python 🏠 Home Appendix C: PySpark →