Building AI APIs and Backends with FastAPI

Why FastAPI for AI backends intermediate

Here's the moment this chapter exists for. You've written Claude code that works beautifully in a notebook — a prompt, a call, a clean answer. Then a teammate asks, "Can my app use it?" and you realise a notebook cell isn't something a phone, a website, or another service can talk to. Your logic needs a front door: a URL that accepts a request over the network and hands back a reply. That front door is an API, and the framework you build it with decides whether your service is fast, safe, and pleasant to run — or a fragile thing that falls over the first busy afternoon.

So why this framework and not Flask or Django, which have been around longer? Because an LLM backend has an unusual personality, and FastAPI was built for exactly it. Picture the difference between a chef and a waiter. A CPU-heavy service is like a chef: it's doing work the whole time — chopping, cooking — so it needs a whole kitchen station per order. An LLM service is like a waiter: it takes your order, walks it to the kitchen (the model), and then just waits. A single waiter can juggle a dozen tables precisely because most of the job is standing around waiting for food to be ready. Your endpoint spends almost all of its time waiting on a slow, token-metered upstream — so the framework you want is one that lets a single worker wait on many requests at once. That's FastAPI's native mode, and it's the whole reason it fits.

The common mistake here is treating "which web framework" as a matter of taste, like tabs versus spaces. It isn't. If you pick a synchronous-first framework and then make blocking model calls, each request hogs a worker for the entire multi-second wait, and your throughput collapses under a load that FastAPI would shrug off. The three things an LLM endpoint needs most — wait efficiently (async), validate and shape data (the Pydantic models you already write for structured output), and stream partial results — are precisely the three things FastAPI makes effortless. You're not learning a new discipline; you're pointing the typing and async habits you already built earlier in the course at an HTTP surface. The table below lines each LLM-backend need up with the FastAPI feature that answers it, so you can see the fit directly.

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.

Lab B3.1 · A typed endpoint intermediate

The first API you write always feels like it must be doing something magical, and it helps to see that it isn't. An endpoint is just an ordinary Python function with a sticky note on it. The sticky note — a decorator — says "hey framework, when a request arrives at this URL, call me." Everything else you'd normally hand-wire for a web service (reading the request body, parsing JSON, checking the fields are the right types, turning your return value back into JSON, writing the docs) FastAPI does for you, and it figures out how by reading the type hints you already know how to write. That's the single idea to hold onto in this lab: you describe the shapes, FastAPI does the plumbing.

Here's the mental model. Think of a Pydantic model as a form at a government office. The form lists exactly which boxes must be filled in and what goes in each — "Question: text. Max tokens: a whole number, and if you leave it blank we'll write 512." When a request shows up, FastAPI checks it against the form before your code sees it. A request that fails the form — a number where text belongs, a missing required box — is bounced at the counter with a 422 error that spells out exactly what was wrong. Your handler only ever runs on requests that already passed inspection, which is why so much defensive code you'd write in other frameworks simply disappears here.

The common beginner mistake is to skip the response model and just return a raw dict, reaching for the typed model only on the way in. It works, but you give up half the benefit: the return type is what lets FastAPI guarantee the JSON you send back has a stable, documented shape, and it's what fills in the response half of the interactive docs. Declare both — request model in, response model out — and the endpoint's signature becomes the contract: anyone (a teammate, a frontend, future-you) can read the two models and know precisely what to send and what they'll get, without reading the body of the function at all. The lab below is the smallest complete example of that contract, echoing the question straight back so the wiring is all you're looking at.

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}")

Lab B3.2 · Async endpoints for LLM calls intermediate

Picture a launderette to feel what async buys you. You have one attendant and ten washing machines. If the attendant loads machine one and then stands there watching it spin for forty minutes before touching machine two, the place is a disaster — nine machines idle, a queue out the door. Obviously that's not how it works: the attendant loads a machine, walks off to load the next while the first runs, and circles back only when a load is done. The machines do the waiting; the attendant stays busy dealing with whatever's ready. An async web server is that attendant, and each slow LLM call is a machine spinning on its own.

This matters for LLM backends specifically because of where the time goes. A model call might take three seconds, but your server does almost no work during those three seconds — it fires off a network request and then waits for tokens to come back. That's called I/O-bound work: bound by input/output (the network), not by the CPU. The magic word is await. When your code hits await client.messages.create(...), it's telling the event loop "I'm going to be waiting a while — go do something useful and wake me when the answer arrives." That handoff is what lets a single worker keep dozens of requests in flight at once. Without it, each request would monopolise the worker for its whole three-second wait, and a handful of simultaneous users would be enough to make everyone else queue.

The trap — and it is the single most common FastAPI-for-LLM bug, worth burning into memory now — is putting blocking code inside an async def route. If you call the synchronous SDK, or time.sleep(), or crunch a big loop, you never hand control back to the event loop. Our launderette attendant is now standing frozen in front of one machine again, and every other request stalls behind them. The rule is simple: in an async def route, everything slow must be await-ed (use the async client), and genuinely blocking work must be pushed to a threadpool so the loop stays free. Async isn't faster for one request — that call still takes its three seconds — it's dramatically better under concurrency, which is exactly the condition a real backend lives in. The diagram below shows one worker holding three awaiting requests at once; keep the launderette in mind as you read it.

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.

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)

llm_endpoint.pyfrom openai import AsyncOpenAI
client = AsyncOpenAI()      # async client — awaitable calls

@app.post("/ask")
async def ask(req: AskRequest) -> AskResponse:   # async def!
    resp = await client.responses.create(
        model="gpt-5.5", max_output_tokens=req.max_tokens,
        input=req.question,
    )
    text = resp.output_text
    return AskResponse(answer=text)

Lab B3.3 · Streaming tokens advanced

Think about why a good chatbot feels alive and a bad one feels broken. Ask ChatGPT or Claude a question and the answer starts appearing almost instantly, word by word, like someone typing to you. Now imagine the same answer arriving as a blank spinner for eight seconds, then a wall of text dumped all at once. The total time is identical — the model took just as long either way — but the first feels responsive and the second feels dead. That gap is entirely about when the user first sees something, and closing it is what this lab is for.

The idea is called streaming, and the honest framing is important: streaming does not make generation faster. The model produces tokens at the same rate regardless. What streaming changes is latency you can perceive — the "time to first token" — by shipping each piece of the answer to the browser the instant the model produces it, instead of buffering the whole reply and sending it in one lump at the end. It's the difference between a waiter bringing each dish as it's cooked versus making you wait until the entire table's order is plated. Same kitchen speed; wildly different experience.

Mechanically, this rests on a Python idea you've met before: the generator. A normal function computes everything and returns once. A generator yields values one at a time, pausing between each — perfect for "hand me the next chunk as it's ready." FastAPI's StreamingResponse takes such a generator and flushes each yielded piece down the open HTTP connection immediately, rather than waiting for the function to finish. So the pattern is: an async generator loops over the model's token stream and yields each chunk; StreamingResponse pipes those chunks to the client live. The subtle mistake to avoid is accidentally re-buffering — for instance, collecting the chunks into a list "to return at the end" — which quietly throws away the entire benefit and turns your stream back into a lump. If you're yielding, yield straight through. There's a real second win here too, covered in the tip below: a long non-streaming response can trip HTTP idle timeouts, whereas a stream keeps bytes flowing and sidesteps that failure mode. The lab shows the whole shape in a handful of lines.

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.

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")

stream_endpoint.pyfrom fastapi.responses import StreamingResponse

@app.post("/ask/stream")
async def ask_stream(req: AskRequest):
    async def token_gen():
        async with client.responses.stream(
            model="gpt-5.5", max_output_tokens=req.max_tokens,
            input=req.question,
        ) as stream:
            async for event in stream:
                if event.type == "response.output_text.delta":
                    yield event.delta   # each chunk flushed to the client
    return StreamingResponse(token_gen(), media_type="text/plain")

Lab B3.4 · Errors, validation & dependencies advanced

The gap between a demo and a service is almost entirely about what happens when things go wrong. Your happy-path endpoint works great when the request is well-formed, the key is valid, and the model answers cleanly. But the moment you expose it to the real world, you get empty questions, forty-thousand-character prompts pasted from a PDF, requests with no auth header, and a model that occasionally rate-limits you or refuses to answer. A demo throws a raw stack trace and a 500 at every one of these. A production endpoint gives each its own honest, correct HTTP status — and this lab wires up the three mechanisms that get you there: validation, dependencies, and error mapping.

Take them one at a time. Validation you've already half-built: a Pydantic Field lets you attach rules right to the model — min_length, max_length, ranges — and FastAPI enforces them before your handler runs, bouncing bad input with a 422 for free. Capping input size isn't just tidiness; an unbounded prompt is a cost-and-latency bomb, since you pay per token. Dependencies are the piece worth slowing down on. Think of Depends as a checklist stapled to the door: "before you let this request in, run these checks and gather these things." An auth check, a database handle, the model client — each becomes a small function FastAPI runs first and injects into your handler. That indirection feels like overhead until you write a test: because the client arrives through Depends, you can swap in a fake one with a single line and test your endpoint without ever touching the real API. Injected seams are testable seams.

Error mapping is the third habit, and the mental model is a translator standing between two worlds. The SDK (Anthropic's or OpenAI's) speaks in Python exceptions — RateLimitError, and so on — while HTTP clients speak in status codes. Your job is to translate: a rate-limit becomes a 503 "busy, retry" (a temporary condition the caller can back off from), a model refusal becomes a 422 "can't answer that" (the request was understood but won't be fulfilled), and only a genuinely broken server should ever surface a 500. The classic mistake is letting SDK exceptions bubble up uncaught, so every upstream hiccup looks to your callers like your service crashed — they can't tell "try again in a second" apart from "this is fundamentally broken," and they retry the wrong things. Catch the exceptions you expect, map each to the status code that tells the caller what to do, and your endpoint becomes something other systems can build on reliably. The lab combines all three on one route.

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

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"))

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

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.responses.create(model="gpt-5.5",
            max_output_tokens=512, input=req.question)
    except openai.RateLimitError:
        raise HTTPException(status_code=503, detail="busy, retry")  # map to HTTP (O3)
    if resp.status != "completed":                      # incomplete / refusal (C2)
        raise HTTPException(status_code=422, detail="cannot answer")
    return AskResponse(answer=resp.output_text)

From app to deployed service advanced

A running app on your laptop and a deployed service look identical in the browser — and could not be more different in reality. The gap is everything that separates "it works when I press play" from "it keeps working at 3am under load, when a machine dies, when the API key rotates, and when a hundred users arrive at once." The reassuring news is that you already learned this material in O3; the point of this section is to show that a FastAPI app is the thing O3 was preparing you to ship. There's no new framework to learn — just the same app, wearing its deployment clothes.

Two ideas are worth calling out because they surprise people coming from CPU-heavy services. First, scale on concurrency, not cores. You met the reason back in the async section: LLM work is I/O-bound, so a single async worker holds many in-flight requests while they wait on the model. That flips the usual autoscaling instinct — you don't add a CPU core per request, you watch the number of concurrent requests and add workers when that climbs. Second, the health check deserves a moment of thought, because the obvious implementation is a trap. Your orchestrator (Kubernetes, a load balancer) pings a /health route constantly to decide if the instance is alive. If that route calls the model, you've turned a liveness probe into a firehose of paid API calls, and a brief model outage will make your orchestrator kill perfectly healthy instances. The right move is a cheap route that returns {"status": "ok"} without touching the model — it answers "is this process up?", which is all a liveness check should ask.

The rest is the boring, load-bearing plumbing: the API key comes from an environment variable injected at runtime, never baked into the image or committed to git (a leaked key is a bill and a breach); the container runs uvicorn to serve the app; graceful shutdown lets in-flight requests finish when a pod is told to stop rather than severing them mid-answer; and middleware logs each request so you can actually see what your service is doing in production. None of these is glamorous, and all of them are the difference between a service you trust and one you babysit. The table maps each O3 concept onto its concrete FastAPI form so the translation is explicit.

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

Serving an agent (not just a call) expert

Everything so far assumed a quick, one-shot call — and an agent breaks that assumption in every direction. A single Claude call is like ordering a coffee: you ask, you wait a few seconds, you get one answer, done. An agent is like hiring a contractor to renovate a room: it works in a loop, calls tools, makes decisions, occasionally does something expensive or irreversible, and might take minutes rather than seconds. You can serve it through a FastAPI endpoint — that part doesn't change — but the naive request/response contract that fit a coffee order fits a renovation badly, and this section is about the three ways the design has to bend.

The first bend is time. An HTTP client holding a connection open for five minutes is asking for trouble — timeouts, dropped connections, a spinner the user assumes is frozen. So a long agent run wants one of two shapes instead of a plain wait: stream the run as progress events (the same streaming muscle from the earlier lab, now carrying "step 3: searching…" instead of answer tokens), or accept the task, immediately return a job id, and let the client poll or receive a webhook when it's done. This is the identical long-running-task pattern you'll see in A2A (I2) and LangGraph's durable state (L5); it's not a FastAPI quirk, it's how any system fronts work that outlives a single request.

The second bend is state. A single call is stateless — each request stands alone. A conversation isn't: turn five needs to remember turns one through four. Since your endpoint might be one of many workers behind a load balancer, you can't stash that history in a local variable and hope the same worker handles the next turn. The caller passes a thread or session id, and the conversation state lives in a shared store or a LangGraph checkpointer (L5) that any worker can read. The third bend is safety and cost, and it's the one that bites hardest: an agent can loop, and a looping agent burns tokens and can take real actions. So the endpoint's job isn't just to run the agent but to bound it — a max-steps or token budget so a runaway can't rack up a fortune — and to make it observable by returning structure, not just a blob of text. The mistake to avoid is fronting an agent exactly like you fronted a single call: block for minutes, return a bare string, and hope. Instead, hand back a typed result carrying what the agent did — the answer, how many steps it used, which tools it called, why it stopped — so the caller can see, debug, and trust it. The tables below lay out each concern and its approach.

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.

Common pitfalls expert

Exercises expert

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

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}

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")

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)}

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"}

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)

Knowledge check check yourself