Data & App Building · Chapter B3

Building AI APIs and Backends with FastAPI

Your LLM logic needs a front door — an HTTP endpoint your app, teammates, or other services can call. FastAPI is the Python standard for that: async-native (perfect for I/O-bound LLM calls), Pydantic-typed (the schemas you already use), and it streams. This chapter turns your Claude code into a real backend.

⏱️ ~65 min🧪 4 labs🎯 Intermediate

⚙️ To run this for realThe code here is complete and correct as written. To actually execute it you'll need:

an Anthropic API key (ANTHROPIC_API_KEY) + pip install anthropic

Reading and learning works without any of this — run when you're ready. IDs/ARNs in examples are placeholders; swap in your own.

Learning objectives

Build a FastAPI app with typed request/response models (Pydantic).
Use async def endpoints correctly for I/O-bound LLM work.
Stream tokens to the client with a streaming response.
Handle errors, validation, and dependencies (auth, clients) cleanly.
Wrap a Claude call — or an agent — as a production-shaped endpoint.

Everything you need is already in the courseFastAPI leans on things you know: Pydantic models (Ch 2, P4), async (A3), streaming the Claude response (C2), error/stop-reason handling (C2/A6), and it's the container you'll deploy in O3. This chapter assembles them into an HTTP service. It's the "front door" for the LLM code you've written all course.

Why FastAPI for AI backends intermediate

An LLM backend has a specific shape — it mostly waits on a slow, token-metered upstream (O3), returns structured data, and often streams. FastAPI fits that shape better than the older Python frameworks.

LLM backend need	FastAPI answer
I/O-bound: mostly waiting on the model	Async-native — one worker serves many concurrent requests (A3, O3)
Structured, validated I/O	Pydantic models validate requests & shape responses (Ch 2)
Stream tokens to the UI	First-class streaming responses (C2)
Clients need to know the contract	Auto-generated OpenAPI docs at `/docs`
Fast to write, easy to test	Type-hint-driven; a route is just a function

The type hints do double dutyFastAPI reads your Python type hints and Pydantic models to (1) validate incoming requests, (2) serialize responses, and (3) auto-generate interactive API docs. The typing discipline you built for structured output (Ch 2) and Pydantic (P4) directly powers a self-documenting, self-validating API — you write types once, get three features free.

Lab B3.1 · A typed endpoint intermediate

Example code for learning — review, test, and adapt it before running against real or production systems. Commands can create, change, or delete resources. See the Terms & Disclaimer.

Lab B3.1

shellpip install "fastapi[standard]"        # includes uvicorn
fastapi dev main.py                     # dev server + auto-reload; docs at /docs

main.pyfrom fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class AskRequest(BaseModel):     # request schema — validated automatically
    question: str
    max_tokens: int = 512

class AskResponse(BaseModel):    # response schema — shapes the JSON out
    answer: str

@app.post("/ask")
def ask(req: AskRequest) -> AskResponse:
    return AskResponse(answer=f"You asked: {req.question}")

▶ How this works

This is the smallest possible FastAPI app: it defines what a request looks like, what a response looks like, and one endpoint (a URL your code answers). FastAPI turns a plain Python function into a working web API for you.

app = FastAPI() creates the application object. Everything else attaches to it.
class AskRequest(BaseModel) is a Pydantic model describing the JSON a caller must send: a question (text) and an optional max_tokens that defaults to 512. FastAPI uses this to validate the incoming request automatically — bad input is rejected before your code runs.
class AskResponse(BaseModel) describes the JSON you send back: just an answer string. This shapes the output into clean, predictable JSON.
@app.post("/ask") is a decorator that says "when someone sends a POST request to the URL /ask, run the function below". def ask(req: AskRequest) -> AskResponse: declares that the body arrives as an AskRequest and the reply is an AskResponse — the type hints are the API contract.
The body just echoes the question back inside an AskResponse. FastAPI converts that object into JSON for the caller.

What the output means: Start the server with fastapi dev main.py, POST {"question": "hi"} to /ask, and you get back {"answer": "You asked: hi"}. Visit /docs and there's a full interactive API page you never had to write.

Try this: Open /docs in a browser and send a test request from there. Then POST a request with max_tokens set to a word instead of a number — FastAPI replies with an automatic 422 error explaining exactly what's wrong.

A route is a typed functionDeclare req: AskRequest and FastAPI parses+validates the JSON body into that model (a malformed request gets an automatic 422 with a clear error). Declare -> AskResponse and it serializes the return value to JSON matching that schema. Open /docs and there's an interactive, correct API reference you didn't write. That's the whole appeal: the contract is the code.

Lab B3.2 · Async endpoints for LLM calls intermediate

This is where FastAPI earns its place for AI. An LLM call spends seconds waiting (O3). With async def and the async Anthropic client, one worker handles many in-flight requests instead of blocking on each — the concurrency lesson from A3, applied to your API.

Waiting concurrently, not serially. Because the work is I/O-bound (idle while the model thinks), an async worker parks each awaiting request and serves the others — so one process handles many concurrent LLM calls. This is why FastAPI + async is the right shape for an LLM backend, and why you don't need a core per request (O3).

🗺️ How to read this diagram

This picture explains why async matters for an LLM backend. A single worker (one process) is serving three requests at the same time — because each one is just waiting on the model, the worker doesn't need to finish one before starting the next.

The box on the left ("1 worker") is a single server process. Normally you might expect it to handle one request at a time.
The three arrows fan out to req A, req B, req C — three requests that arrived at nearly the same moment. Each label says "awaiting model": the request has been sent to Claude and is now doing nothing but waiting for the answer.
The box on the right ("Claude") is the slow upstream — the language model the requests are all waiting on. That wait (often seconds) is the whole reason this works.
The key idea: while req A is parked waiting, the one worker is free to start req B, then req C. It overlaps the waiting instead of standing idle. This is I/O-bound work — mostly waiting, not computing — which is exactly what async is good at.

In short: One async worker can hold many in-flight LLM calls at once because the calls spend their time waiting, not using the CPU. That's why you don't need one CPU core per request.

Lab B3.2

Requires: pip install anthropic

llm_endpoint.pyfrom anthropic import AsyncAnthropic
client = AsyncAnthropic()      # async client — awaitable calls

@app.post("/ask")
async def ask(req: AskRequest) -> AskResponse:   # async def!
    resp = await client.messages.create(
        model="claude-opus-4-8", max_tokens=req.max_tokens,
        messages=[{"role":"user","content": req.question}],
    )
    text = next(b.text for b in resp.content if b.type == "text")
    return AskResponse(answer=text)

▶ How this works

Now the endpoint actually calls Claude. The important change from Lab B3.1 is the two words async and await — they are what let one worker serve many slow LLM requests at once (the diagram above).

client = AsyncAnthropic() creates the async version of the Anthropic client. Its calls are awaitable: you can pause on them without freezing the whole server.
async def ask(...) — the word async marks this as a coroutine. While it's paused waiting, FastAPI can run other requests. A plain def here would block everyone (see the warning box below).
resp = await client.messages.create(...) sends the question to the model and await means "pause here until the reply comes back, but let other requests run meanwhile". The model, max_tokens, and messages are the same request shape you learned in the earlier Claude chapters.
next(b.text for b in resp.content if b.type == "text") pulls the first text block out of the reply (the response is a list of blocks, not a plain string), and we return it wrapped in AskResponse.

What the output means: A POST to /ask now returns Claude's real answer as {"answer": "..."}. Under load, one worker handles many of these at the same time because each is mostly waiting.

Try this: Fire several requests at once (open /docs in a few tabs and submit together). Because of async/await, they overlap instead of lining up one behind the other.

Don't block the event loopIn an async def route, a synchronous blocking call (a sync SDK call, time.sleep, heavy CPU work) freezes the whole worker — every concurrent request stalls (A3). Use the async client and await it. If you must call blocking code, run it in a threadpool (FastAPI does this automatically for plain def routes, or use run_in_threadpool). Mixing a sync SDK call into an async route is the #1 FastAPI-for-LLM mistake.

Lab B3.3 · Streaming tokens advanced

For a chat UI, you want tokens to appear as they're generated, not after a multi-second wait (C2, O3). FastAPI streams via a streaming response that yields chunks.

Lab B3.3

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

stream_endpoint.pyfrom fastapi.responses import StreamingResponse

@app.post("/ask/stream")
async def ask_stream(req: AskRequest):
    async def token_gen():
        async with client.messages.stream(
            model="claude-opus-4-8", max_tokens=req.max_tokens,
            messages=[{"role":"user","content": req.question}],
        ) as stream:
            async for text in stream.text_stream:
                yield text          # each chunk flushed to the client
    return StreamingResponse(token_gen(), media_type="text/plain")

▶ How this works

This endpoint sends the answer back word-by-word as the model writes it, instead of waiting for the whole reply. That's what makes a chat UI feel like live typing rather than a long freeze then a wall of text.

@app.post("/ask/stream") defines a second endpoint at /ask/stream so you keep the plain /ask and add a streaming one alongside it.
async def token_gen(): is an inner generator — a function that hands back pieces one at a time using yield, rather than returning everything at once.
async with client.messages.stream(...) as stream: opens a streaming connection to Claude. The async with makes sure that connection is closed cleanly when the reply is finished.
async for text in stream.text_stream: receives small chunks of text as the model produces them, and yield text immediately passes each chunk onward to the caller — nothing is held back.
return StreamingResponse(token_gen(), media_type="text/plain") hands FastAPI the generator. FastAPI then streams each yielded chunk to the client as it appears.

What the output means: The caller sees text arrive progressively, a few words at a time, instead of one big response after several seconds. The user gets feedback almost instantly.

Try this: Call /ask/stream with curl -N http://localhost:8000/ask/stream (the -N disables buffering) and watch the answer type itself out live.

Streaming solves latency and timeouts at onceStreaming gives the user immediate feedback (time-to-first-token, not total time) and avoids the HTTP timeout risk of a long non-streaming response (O3). For a richer client contract, format chunks as Server-Sent Events (SSE) — the same wire format the Anthropic API itself streams (C2). The generator pattern (async def + yield) is the async version of A3's streaming.

Lab B3.4 · Errors, validation & dependencies advanced

A production endpoint validates input, handles the LLM's failure modes, and manages shared resources (the client, auth) cleanly.

Lab B3.4

Uses the client / objects set up earlier in this lesson, and needs ANTHROPIC_API_KEY set. Run the earlier blocks first.

robust.pyfrom fastapi import FastAPI, HTTPException, Depends, Header
from pydantic import BaseModel, Field
import anthropic

class AskRequest(BaseModel):
    question: str = Field(min_length=1, max_length=4000)   # validation for free

def require_key(x_api_key: str = Header(...)):        # a dependency: auth
    if not _valid(x_api_key):
        raise HTTPException(status_code=401, detail="invalid key")

@app.post("/ask", dependencies=[Depends(require_key)])
async def ask(req: AskRequest) -> AskResponse:
    try:
        resp = await client.messages.create(model="claude-opus-4-8",
            max_tokens=512, messages=[{"role":"user","content": req.question}])
    except anthropic.RateLimitError:
        raise HTTPException(status_code=503, detail="busy, retry")  # map to HTTP (O3)
    if resp.stop_reason == "refusal":                    # handle stop reasons (C2)
        raise HTTPException(status_code=422, detail="cannot answer")
    return AskResponse(answer=next(b.text for b in resp.content if b.type=="text"))

▶ How this works

A demo endpoint becomes a production endpoint when it checks its input, requires authentication, and turns the model's failures into proper HTTP errors instead of crashing. This block adds all three.

question: str = Field(min_length=1, max_length=4000) — the Pydantic Field adds rules to the input. An empty or too-long question is rejected automatically with a 422 before your code even runs — validation you get for free just by declaring it.
def require_key(x_api_key: str = Header(...)) is a dependency: a small check FastAPI runs first. It reads an x-api-key HTTP header and, if the key is invalid, raises HTTPException(status_code=401, ...) — the standard code for "not authenticated".
@app.post("/ask", dependencies=[Depends(require_key)]) attaches that auth check to the endpoint. Depends() is how FastAPI wires in shared pieces (auth, a database, the client) so they're reusable and easy to test.
The try/except wraps the Claude call: a RateLimitError (the API is busy) is turned into a 503 "busy, retry" instead of a stack trace. This maps an SDK error to a sensible HTTP code.
if resp.stop_reason == "refusal": handles the case where the model declined to answer, returning a 422 "cannot answer" rather than pretending it succeeded.

What the output means: Bad input → automatic 422. Missing/invalid key → 401. Model busy → 503. Model refuses → 422. Only a genuinely successful call returns the answer — every failure mode has a clear, correct HTTP status.

Try this: Send a request with no x-api-key header and confirm you get a 401. Then send a question longer than 4000 characters and watch the 422 from the Field limit.

Concern	FastAPI mechanism
Input validation	Pydantic `Field` constraints → automatic 422 on bad input
LLM failures	Catch SDK exceptions, map to sensible HTTP codes (429→503, refusal→422) — O3 degradation
Shared resources	`Depends()` injects the client, DB, or auth — testable & reusable
Auth	A dependency that checks a header/token; never hardcode keys (T1)
Secrets	API key from env, injected at runtime (O3, T1) — not in code

Your endpoint is now an attack surface (T1)Exposing an LLM over HTTP means everything from Topic T1 applies: prompt injection via the request body, abuse/cost-blowup from unauthenticated access, and data leaks in responses. Authenticate, rate-limit, validate input size, put guardrails on I/O (I5), and never echo secrets. A public LLM endpoint with no auth is a bill and a breach waiting to happen.

From app to deployed service advanced

A FastAPI app is exactly what O3 taught you to containerize and scale. The connection is direct:

O3 concept	Here it is in FastAPI
Containerize with runtime secrets	The Dockerfile from O3 runs `uvicorn main:app`; key via env
Scale on concurrency, not CPU	Async workers + autoscale on request concurrency
Health check	A cheap `@app.get("/health")` that doesn't call the model
Graceful shutdown	Uvicorn drains in-flight (slow) requests on SIGTERM
Observability	Middleware logs the four signals per request (O4, I4)

This is the backend the whole course points atThe projects (support agent, doc-intel, data analyst) all need an HTTP front door — this is it. Wrap your agent's run() in an endpoint, add the health check and auth, containerize per O3, and monitor per O4. FastAPI is where your LLM logic becomes a service other systems can actually call.

Serving an agent (not just a call) expert

The endpoints above wrap a single Claude call, but the same shape serves a whole agent (Ch 4, L4/L5). One caveat: agent runs are long and stateful, which changes the endpoint design.

Agent-serving concern	Approach
Runs take a long time	Stream progress, or return a job id and poll/webhook — don't block for minutes
Stateful conversations	Pass a thread/session id; back state with a store or LangGraph checkpointer (L5)
Risky actions	Human-approval steps become endpoints too (approve/reject) — the L5 interrupt over HTTP
Cost/abuse	Auth, rate limits, per-user budgets at the API edge (O2 gateway)

Long-running work needs a different contractA 5-second call fits request/response; a 5-minute agent run does not. Either stream the whole run (progress events), or accept the task and return a job id the client polls — the same long-running-task shape as A2A (I2) and LangGraph's durable state (L5). Don't make an HTTP client hold a connection open for minutes.

Common pitfalls expert

Pitfall	Fix
Sync SDK call inside an `async def` route	Use the async client + `await`; or a threadpool
Blocking the event loop (sleep, heavy CPU)	Offload to a threadpool/worker; keep routes non-blocking
No input validation	Pydantic `Field` constraints; cap sizes
Unhandled LLM errors/stop reasons	Catch SDK exceptions; branch on `stop_reason` (C2)
Blocking for a long agent run	Stream, or job-id + poll/webhook
Unauthenticated public endpoint	Auth dependency, rate limits, guardrails (T1, I5)
Health check that calls the model	Cheap liveness route; probe the model sparingly (O3)

Exercises expert

Exercise B3.1 — Typed classify endpoint

Context: The classifier from Chapter 2 becomes a real service the moment it has a typed contract and input limits. FastAPI's auto-docs and 422s do the tedious guarding for you.

Your task: Build a POST /classify endpoint that takes text and returns a typed {"label", "confidence"} using the Chapter 2 classifier, with a Pydantic Field capping input length.

Requirements:

Return a typed response with a label and a confidence
Use a Pydantic Field to cap the input text length
Confirm a too-long request gets an automatic 422 — no manual check
Verify the endpoint appears in the auto-generated docs at /docs

💡 Hint: Put the length cap in the request model's Field so the 422 is generated for you, then wire the classifier into the response model.

Exercise B3.2 — Sync vs async under load

Context: The async payoff is invisible until you put concurrency on it. Firing many requests at a sync endpoint and an async one side by side makes the event-loop story from the async lesson measurable.

Your task: Write the same LLM endpoint two ways — a plain def with the sync client and an async def with the async client — fire ~20 concurrent requests at each, compare total time, and explain the gap.

Requirements:

Implement the endpoint both as sync (def + sync client) and async (async def + async client)
Fire roughly 20 concurrent requests at each version
Measure and compare total wall-clock time for both
Explain the difference in terms of the event loop overlapping I/O waits
Show the async version finishing far faster under concurrency

💡 Hint: The async version overlaps the waiting while the blocking version serializes — the concurrent load is what exposes the difference.

Show what to look for

The async version finishes far faster under concurrency because one worker overlaps the waiting; the sync-in-async version (or a blocking call) serializes and stalls. This is the I/O-bound story from O3 made measurable — and why the async client matters.

Exercise B3.3 — Stream + harden

Context: A deployable service is the streaming endpoint plus the boring hardening: auth, error mapping, and a health route that never touches the model. That's what makes it O3-ready rather than a demo.

Your task: Add a streaming /ask/stream endpoint and an auth dependency, map a rate-limit error to a 503 and a refusal to a 422, and add a /health route that doesn't call the model.

Requirements:

Provide a streaming /ask/stream endpoint
Gate requests with an auth dependency
Map a rate-limit error to a 503 and a model refusal to a 422
Add a /health route that returns without calling the model
The result is a deployable, O3-ready service

💡 Hint: Reuse the streaming generator from the ladder, add the auth via Depends, and translate each upstream failure into its own status code.

🪜 Practice ladder beginner → industry

Six graded exercises, easy to real-world. Try each before opening its solution.

Exercise 1 · A typed POST endpoint with a Pydantic modelBeginner

Context: In FastAPI the request model is the contract, and typing it buys you validation and auto-generated docs for free. A malformed request is rejected before your handler ever runs.

Your task: Write a FastAPI POST /echo endpoint that accepts {"text": str} via a Pydantic model and returns the text's length, then show why the model gives you validation for free.

Requirements:

Define a Pydantic BaseModel with a required text: str field
The endpoint takes the model as its body and returns the length of the text
A request with the wrong type (e.g. a number) is auto-rejected with a 422 — no hand-written check
Note the run command (uvicorn main:app) and that validation is a side effect of typing

💡 Hint: Just declaring the field type on the model is the whole validation story; FastAPI does the 422 for you before your function body executes.

Show solution

Needs fastapi + uvicorn to serve. The request model is the contract; FastAPI validates and documents it automatically:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class EchoIn(BaseModel):
    text: str                     # required, must be a string

@app.post("/echo")
def echo(body: EchoIn):
    return {"length": len(body.text)}

# run:  uvicorn main:app --reload
# POST {"text": "hi"}   -> {"length": 2}
# POST {"text": 123}    -> 422 auto-validation error, no code needed

Declaring text: str means malformed requests are rejected with a 422 before your code runs — validation is a side effect of typing the model.

Exercise 2 · Make the LLM call async so the server stays responsiveIntermediate

Context: An LLM call is I/O-bound — it spends most of its time waiting on the network. Handled with async def and await, that waiting frees the event loop to serve other requests instead of blocking a worker.

Your task: Explain why async def plus await matters for a web server calling a slow LLM, and write the async endpoint shape that awaits the model call.

Requirements:

Explain that the LLM call is I/O-bound and await yields the event loop while waiting
The endpoint is declared async def and takes a typed Pydantic body
It awaits the async model call rather than blocking
State the rule: never do blocking work inside an async def or you stall the whole loop
The shape is complete enough to serve many concurrent users from one process

💡 Hint: Mark the handler async def and await the network call — the win is that one process overlaps many requests' waiting.

Show solution

Why async: an LLM call is I/O-bound — it spends most of its time waiting on the network. With async def + await, that waiting frees the event loop to handle other requests instead of blocking a worker, so one process serves many concurrent users.

Needs fastapi + an async client. Endpoint shape:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class AskIn(BaseModel):
    prompt: str

async def call_model(prompt: str) -> str:
    # await the async LLM client here; returns the model's text
    ...

@app.post("/ask")
async def ask(body: AskIn):
    answer = await call_model(body.prompt)   # yields the loop while waiting
    return {"answer": answer}

Use await on the network call; never do blocking work inside an async def or you stall the whole event loop.

Exercise 3 · Stream tokens back to the clientAdvanced

Context: Streaming doesn't make generation faster, but it slashes perceived latency: the user sees the first tokens almost immediately instead of staring at a spinner for the whole answer.

Your task: Return a streaming response so the client receives tokens as they're generated, using an (async) generator wrapped in a StreamingResponse.

Requirements:

Write an (async) generator that yields each token/delta as it arrives
Wrap that generator in a StreamingResponse
Set an appropriate media_type (e.g. text/plain)
The endpoint returns the streaming response rather than a fully-built string
Note that this improves time-to-first-token, not total generation time

💡 Hint: The generator is the engine and StreamingResponse is the wrapper — in real code you iterate the model's streaming API and yield each delta.

Show solution

Needs fastapi. Yield chunks from an (async) generator and wrap it in StreamingResponse:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def token_stream(prompt: str):
    # in real code, iterate the model's streaming API and yield each delta
    for tok in ["Hel", "lo, ", "world"]:
        yield tok

@app.post("/stream")
async def stream(prompt: str):
    return StreamingResponse(token_stream(prompt), media_type="text/plain")

Streaming improves perceived latency — time-to-first-token drops sharply even if total generation time is unchanged. The generator yields each delta as it arrives.

Exercise 4 · Errors, validation, and a dependencyExpert

Context: Production endpoints need to reject bad input with a clean status code and get their shared resources injected rather than built inline. HTTPException and Depends are how FastAPI does both without a mess of globals.

Your task: Add real-world robustness to an ask endpoint: reject an empty prompt with a clean 400 and inject a shared resource (like a model client) via Depends.

Requirements:

Validate the prompt and raise HTTPException(status_code=400, ...) when it's empty
Provide the shared client through a dependency function used with Depends
The endpoint receives the injected client as a parameter
The dependency is a swappable seam — a fake can be injected in tests
An expected error surfaces as a proper status code, not an uncaught 500

💡 Hint: Raise HTTPException for the empty-prompt case and let Depends(get_client) hand the client in, which is what makes the endpoint testable.

Show solution

Needs fastapi. Raise HTTPException for expected errors; use Depends so the endpoint gets its dependencies injected (testable, reusable):

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

app = FastAPI()

class AskIn(BaseModel):
    prompt: str

def get_client():
    # build/return a shared model client; swap for a fake in tests
    return {"model": "demo"}

@app.post("/ask")
async def ask(body: AskIn, client=Depends(get_client)):
    if not body.prompt.strip():
        raise HTTPException(status_code=400, detail="prompt must not be empty")
    return {"model": client["model"], "chars": len(body.prompt)}

Depends makes the client a swappable seam (inject a fake in tests), and HTTPException turns an expected error into a proper status code instead of a 500.

Exercise 5 · A health check + validate the pure logic without a serverProfessional

Context: Endpoints need a running server to exercise, but the rules they enforce shouldn't. Pushing validation into a Pydantic model lets you test the contract in milliseconds with no HTTP at all, and add the mandatory /health route.

Your task: Add a GET /health endpoint, and factor the request validation into a Pydantic model with a validator you can unit-test in pure Python without starting a server.

Requirements:

Define a GET /health route (needs FastAPI to serve)
Put the prompt rule in a Pydantic field_validator that rejects blank input
A valid model instance is accepted; a blank one raises ValidationError
Assert both cases in plain Python — no web server required to test the contract
The same validator protects the live endpoint

💡 Hint: The validator lives on the model, so you can construct the model in a test and assert it raises on blank input while the server-bound routes stay separate.

Show solution

The /health endpoint needs fastapi to serve. But the validation logic lives in a Pydantic model you can test with no server at all — this block runs clean:

from pydantic import BaseModel, field_validator, ValidationError

class AskIn(BaseModel):
    prompt: str

    @field_validator("prompt")
    @classmethod
    def not_blank(cls, v):
        if not v.strip():
            raise ValueError("prompt must not be empty")
        return v

# unit test the contract without starting a web server:
assert AskIn(prompt="hi").prompt == "hi"
try:
    AskIn(prompt="   ")
    raise AssertionError("should have rejected blank prompt")
except ValidationError:
    print("blank prompt correctly rejected")

# the health endpoint (needs fastapi to serve):
# @app.get("/health")
# def health(): return {"status": "ok"}

Pushing validation into the model means you test the rules in milliseconds without HTTP — and the same rule protects the live endpoint.

Exercise 6 · Serve an agent, not just a callIndustry scenario

Context: Fronting a multi-step agent is different from fronting a single call: an agent can loop, call tools, and run up cost. The endpoint's real job is to bound it and make its behavior observable.

Your task: Design the production endpoint that fronts a multi-step agent — what it returns, how it bounds cost, and how failures surface — and provide the response shape.

Requirements:

Bound the loop with a max-steps or token budget and return a clear 'gave up' result when exceeded
Return structure, not just text — e.g. answer, steps used, tools called, and a stopped reason
Fail gracefully: wrap tool/model errors as a 502/503 with a request id, never a raw stack trace
Stream progress for long runs so the client isn't left on a spinner
Model the structured result with a Pydantic model (the running system needs the full stack)

💡 Hint: Make the result a Pydantic model carrying stopped_reason ("done" | "max_steps" | "error") so callers can see what the agent did and the budget cap keeps it from running away.

Show solution

Needs the full FastAPI + agent stack to run; the design is the deliverable:

Bound the loop: pass a max-steps / token budget into the agent; return a clear “gave up” result if it’s exceeded rather than looping forever (and running up cost).
Return structure, not just text: {answer, steps_used, tools_called, stopped_reason} so callers can see what happened and debug.
Fail gracefully: wrap tool/model errors and surface a 502/503 with a request id; never leak a raw stack trace.
Stream progress for long runs so the client isn’t staring at a spinner.

from pydantic import BaseModel

class AgentResult(BaseModel):
    answer: str
    steps_used: int
    tools_called: list[str]
    stopped_reason: str            # "done" | "max_steps" | "error"

# endpoint (needs the stack):
# @app.post("/agent")
# async def run_agent(body: AskIn) -> AgentResult:
#     return await agent.run(body.prompt, max_steps=8)

Fronting an agent is different from fronting a call: the endpoint’s job is to bound it (steps/budget) and make its behavior observable (structured result), because an agent can otherwise run away.

✓ Checkpoint — you can move on when you can…

Build a FastAPI app with typed request/response models.
Write an async def LLM endpoint and explain why async fits.
Stream tokens to the client with a streaming response.
Validate input, handle LLM errors/stop reasons, and inject dependencies.
Describe how this app maps to O3 deployment and serving an agent.

🏗️ Toward the capstone & projectsEvery project in the gallery — and the AI DevOps Engineer's control surface — is a FastAPI backend at heart: typed endpoints, async Claude calls, streaming, an auth/guardrail edge, human-approval routes for risky actions (L5), and O4 logging middleware. This chapter is how the agent logic you built becomes a service the world can call. Next, B4 gives it a UI without the HTML. Next: Streamlit & Gradio →

Knowledge check check yourself

✓ Knowledge check

Why is async def the right shape for an LLM endpoint, and what happens if you make a blocking call inside one?

Show answer

LLM work is I/O-bound — mostly waiting on the model — so one async worker can park each awaiting request and serve others concurrently; a synchronous blocking call inside an async route freezes the whole worker and stalls every concurrent request.

✓ Knowledge check

How do FastAPI's Pydantic type hints do "double duty" on an endpoint?

Show answer

FastAPI reads the request/response models to (1) validate incoming requests (bad input gets an automatic 422 before your code runs), (2) serialize responses to matching JSON, and (3) auto-generate the interactive API docs at /docs — you write types once and get three features.

Educational content, provided as-is and without warranty. Code samples are examples — review, test, and adapt them before using in production. See the Terms of Use & Disclaimer. Use at your own risk.

Learning objectives

Why FastAPI for AI backends intermediate

Lab B3.1 · A typed endpoint intermediate

Lab B3.2 · Async endpoints for LLM calls intermediate

Lab B3.3 · Streaming tokens advanced

Lab B3.4 · Errors, validation & dependencies advanced

From app to deployed service advanced

Serving an agent (not just a call) expert

Common pitfalls expert

Exercises expert

Exercise B3.1 — Typed classify endpoint

Exercise B3.2 — Sync vs async under load

Exercise B3.3 — Stream + harden

🪜 Practice ladder beginner → industry

✓ Checkpoint — you can move on when you can…

Knowledge check check yourself

More in Data & app building