Building AI endpoints with FastAPI

Farah’s two lunchtime 500 errors turned out to be the AI provider having a bad minute: two 529 overloaded responses during the lunch rush. Arjun’s endpoint had no idea what to do with them, so FastAPI did the only thing it could. It returned 500 Internal Server Error, the code for “this server crashed”. Farah’s service, seeing a 500, assumed Arjun’s code was broken and stopped sending messages for an hour. About 70 WhatsApp messages sat unclassified.

Meanwhile Meera had a request of her own. “The website chat shows nothing for six or seven seconds, then the whole answer appears at once. Customers keep typing ‘hello??’ into it. Can it type the answer out as it thinks, like ChatGPT does?”

Arjun decided to fix everything in one go. He had read that FastAPI is fast with async, so he changed every def to async def. Then, to be safe, he ran a quick test that fired ten requests at /triage at the same moment.

The first answer came back in 3 seconds. The second in 6. The tenth in 31.

Priya looked at the numbers. “You’ve built a counter where one waiter takes an order, walks it to the kitchen, and stands at the kitchen door until the food is ready. Ten customers, one waiter, standing still. Let’s fix the waiter, then teach him to explain problems politely, then teach him to bring the food out a bite at a time.”

Arjun’s “async” endpoint, and why it got slower

Here is what Arjun had after his change:

import anthropic
from fastapi import FastAPI

app = FastAPI()

@app.post("/triage")
async def triage(req: TriageRequest) -> TicketTriage:
    client = anthropic.Anthropic()            # new SYNC client on every request
    response = client.messages.parse(...)     # blocking call inside async def
    return response.parsed_output

And here is the test script he used, which is worth keeping; it is the cheapest load test you will ever write:

# load_test.py
import asyncio
import time
import httpx

async def one(client: httpx.AsyncClient, i: int):
    start = time.perf_counter()
    r = await client.post("http://127.0.0.1:8000/triage", json={"message": f"Order CR-10{i} is late"})
    print(f"request {i}: {r.status_code} after {time.perf_counter() - start:.1f}s")

async def main():
    async with httpx.AsyncClient(timeout=120) as client:
        await asyncio.gather(*(one(client, i) for i in range(10)))

asyncio.run(main())

Output:

request 0: 200 after 3.1s
request 1: 200 after 6.3s
request 2: 200 after 9.2s
...
request 9: 200 after 31.0s

Each request waited for every request before it. The server was handling exactly one AI call at a time.

You met the reason in the batching lesson. An async def function runs on the event loop, a single thread that juggles many tasks by switching between them at every await. Arjun’s code calls the synchronous client, which has no await in it. While it waits three seconds for Claude, it holds the event loop’s only thread, and no other request can even begin. Ironically, his old plain def endpoint was better: FastAPI runs plain def endpoints in a pool of worker threads, so a few could wait in parallel.

That is the misconception to drop: “async def makes code fast.” It makes code able to wait well, and only if everything slow inside it is awaited.

Riders and phone lines

“Two things are wrong,” Priya said. “The waiter is standing at the kitchen door. And look at the first line of the function.”

client = anthropic.Anthropic().

“Imagine Crumb hired a brand-new delivery rider for every single order, then let them go after one delivery. What’s slow about that?”

“Hiring. Paperwork. They don’t know the roads.”

“A model client is like a rider who knows the roads. Underneath, it keeps a pool of open network connections to the provider. Opening a new secure connection takes a few round trips across the internet before a single byte of your prompt is sent. Create the client once when the app starts, and every request reuses warm connections.”

“So: one shared async client, created at startup, and await every call.”

“That’s fix one.”

Following one request through the fixed endpoint

Here is a single /triage request through the finished code below, with timings from Arjun’s laptop:

Time	Where	What happens
0.000 s	FastAPI	Request arrives; route matched; body validated against TriageRequest (under a millisecond)
0.001 s	Endpoint	await client.messages.parse(...) sends the request over an already-open connection, then yields the event loop
0.001 to 2.9 s	Event loop	Free. Nine other requests start their own model calls in this window
2.9 s	Endpoint	Claude’s reply arrives; the SDK validates it into a TicketTriage object
2.9 s	FastAPI	Checks the object against response_model, serialises it, sends 200

The same ten-request load test against the fixed endpoint:

request 3: 200 after 2.8s
request 0: 200 after 2.9s
...
request 7: 200 after 3.4s

All ten in about 3.4 seconds instead of 31. Same model, same prompts, same bill. Notice that the requests finished out of order: each one returns whenever its own model call does.

Fix 1: async all the way down, one shared client

# main.py
import logging
import os
from typing import Literal

import anthropic
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

load_dotenv()
log = logging.getLogger("crumb.api")
MODEL = os.getenv("CLAUDE_MODEL", "claude-opus-5-5")

# ONE client for the whole app: async, with a timeout and SDK retries.
client = anthropic.AsyncAnthropic(timeout=30.0, max_retries=2)

app = FastAPI(title="Crumb AI API")

class TriageRequest(BaseModel):
    message: str = Field(min_length=1, max_length=4000, description="The customer's message")

class TicketTriage(BaseModel):
    categories: list[Literal["late_delivery", "allergy_question", "refund_request", "other"]]
    urgency: Literal["low", "medium", "high"]
    order_id: str | None = Field(description="Order id like CR-1042, or null")
    summary: str = Field(description="One short sentence in English")

@app.post("/triage", response_model=TicketTriage)
async def triage(req: TriageRequest) -> TicketTriage:
    response = await client.messages.parse(        # await: the event loop is free while we wait
        model=MODEL,
        max_tokens=2000,
        system="You triage Crumb & Co. support messages. Allergy questions are always high urgency.",
        messages=[{"role": "user", "content": req.message}],
        output_format=TicketTriage,
    )
    if response.parsed_output is None:
        # HTTP 200 from the provider, but no usable object (for example a refusal)
        raise HTTPException(status_code=502, detail="The assistant could not classify this message.")
    return response.parsed_output

Everything from this module so far is in this one endpoint: a Pydantic request with a length cap (which is also a cost cap), structured outputs for a typed response, a timeout, SDK retries for the “their busy” and “network” families, and a check for the “success that isn’t”.

The client is created at module level, meaning when Python first imports main.py, which happens once when the server starts. Every request shares it. (For apps that need careful startup and shutdown, FastAPI also offers lifespan functions; module level is fine here.)

Show the answer

No. A blocking call inside async def freezes the event loop, which is exactly what made Arjun’s test take 31 seconds. With a sync-only library, use a plain def endpoint. FastAPI will run it in its thread pool, so several requests can still wait in parallel. Use async def only when everything slow inside it is awaited. (There are also helpers to push one blocking call into a thread from async code, but the plain def endpoint is the simplest correct answer.)

Fix 2: explain problems politely, with the right status code

Now the lunchtime 500s. When an exception escapes an endpoint unhandled, FastAPI has only one honest answer: 500, “this server crashed”. But nothing in Arjun’s server crashed. The provider was overloaded for a minute. The right message to Farah’s service was “busy, try again in 30 seconds”, and her code would have retried instead of giving up for an hour.

FastAPI’s exception handlers let you translate exceptions into HTTP responses in one place. You register a function for an exception class; whenever an exception of that class (or a subclass) escapes any endpoint, FastAPI calls your function. When several handlers could match, the one for the most specific class wins.

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(anthropic.RateLimitError)
async def on_rate_limit(request: Request, exc: anthropic.RateLimitError):
    log.warning("Rate limited on %s", request.url.path)
    return JSONResponse(
        status_code=503,
        content={"error": "The assistant is busy. Please try again shortly."},
        headers={"Retry-After": "30"},
    )

@app.exception_handler(anthropic.APIConnectionError)
async def on_connection_error(request: Request, exc: anthropic.APIConnectionError):
    log.warning("Could not reach the model provider: %s", exc)
    return JSONResponse(status_code=503, content={"error": "The assistant is temporarily unavailable."})

@app.exception_handler(anthropic.APIStatusError)
async def on_provider_error(request: Request, exc: anthropic.APIStatusError):
    log.error("Provider returned %s: %s", exc.status_code, exc)
    if exc.status_code >= 500:
        return JSONResponse(status_code=503, content={"error": "The assistant is temporarily unavailable."},
                            headers={"Retry-After": "30"})
    # A 4xx from the provider means OUR request to it was wrong: a bug on our side, not the caller's.
    return JSONResponse(status_code=502, content={"error": "The assistant failed to respond."})

Here is how the codes map, from the caller’s point of view:

We return	Meaning for Farah’s service
422	Your request was invalid. Fix it; do not retry. (FastAPI does this for you.)
502	Our upstream gave us something we could not use. A human should look; retrying may not help.
503 + Retry-After	We or our provider are busy. Retry after the given number of seconds.
500	We have a genuine bug. Alert us.

With this in place, the lunchtime overload would have produced two 503 responses with Retry-After: 30. Farah’s service, which already honoured Retry-After, would have retried both 30 seconds later and never paused.

Show the answer

No. A 400 tells the caller “your request was wrong”, and Farah’s developer would waste an afternoon hunting for a bug in her code. The bad request was between your server and the provider, so from the caller’s view your server failed. Return 502 (or 500) and log the provider’s message for yourself. The status code should always describe the problem from the caller’s side of the counter.

Fix 3: bring the food out a bite at a time

Now Meera’s request. Streaming means sending the reply in small pieces as the model generates them, instead of waiting for the whole thing. The total time barely changes, but the time until the first words appear drops dramatically, and that is what a human perceives as speed.

Here is one real chat question traced through the streaming endpoint below, with timings from Arjun’s terminal:

Time	What the customer sees
0.0 s	Sends: “What eggless cakes do you have for Diwali?”
0.1 s	Response headers arrive (status 200); the bubble shows a typing indicator
1.2 s	“We have three eggless”
1.4 s	" options this Diwali:"
…	A few words every fraction of a second
5.8 s	“…Shall I check delivery to your area?” Stream ends

Without streaming, the customer stared at nothing for 5.8 seconds. With it, words appeared after 1.2 seconds. Engineers call that first number time to first token, and for chat it matters far more than total time. Meera’s “hello??” messages stopped the day this shipped.

FastAPI’s StreamingResponse sends whatever a generator yields, piece by piece, and the Anthropic SDK’s messages.stream gives you the text as it is generated:

from fastapi.responses import StreamingResponse

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

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    async def generate():
        try:
            async with client.messages.stream(
                model=MODEL,
                max_tokens=2000,
                system="You are Crumb & Co.'s friendly support assistant. Keep answers short.",
                messages=[{"role": "user", "content": req.message}],
            ) as stream:
                async for text in stream.text_stream:   # each piece of text as it is generated
                    yield text
        except anthropic.APIError:
            # Status 200 was already sent with the first byte, so we cannot change it now.
            # The best we can do is end the stream with a polite note, and log the details.
            log.exception("Stream failed")
            yield "\n\n[Sorry, the assistant was interrupted. Please try again.]"

    return StreamingResponse(generate(), media_type="text/plain; charset=utf-8")

generate is an async generator: an async def function that uses yield, producing values one at a time. StreamingResponse pulls from it and writes each piece to the network immediately. Test it with curl -N; the -N flag turns off curl’s own buffering so you see each piece arrive:

curl -N -X POST http://127.0.0.1:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "What eggless cakes do you have for Diwali?"}'

Now look again at the except block, because it hides the most important idea in streaming. An HTTP response starts with its status line. To send the first word, the server must already have sent “200 OK”. If the provider fails three sentences in, there is no way to go back and change it to 503. That is why the exception handlers from Fix 2 cannot help here, and why the try lives inside the generator: errors mid-stream must be signalled inside the stream itself. Module 19 upgrades this to Server-Sent Events, a standard streaming format browsers understand natively, with a proper event type for errors.

Connecting a fuller LLM workflow

Real endpoints usually chain steps. Farah’s next request was “draft a reply, not just a category”. That means: triage the message, look up the order in the database, then ask the model to write a reply using the real order facts. Keep the endpoint thin and put the workflow in a service function:

# services/support.py
async def draft_reply(message: str) -> dict:
    triage_result = await classify(message)                   # model call 1: structured output
    order = await lookup_order(triage_result.order_id)        # your database, not the model
    draft = await write_reply(message, triage_result, order)  # model call 2, given real facts
    return {"triage": triage_result, "draft": draft}

# main.py
@app.post("/support/draft")
async def support_draft(req: TriageRequest):
    return await draft_reply(req.message)

The endpoint knows HTTP. The service knows the workflow. Each model call knows its prompt. Each piece can be tested on its own, and the error handlers from Fix 2 still cover every model call inside, because they catch exceptions from anywhere in the request.

Look at the middle step. The code decides to look up the order, and passes the facts to the model. In Module 06 you will flip this around: the model will decide for itself when it needs to call lookup_order, using function calling.

Where it still breaks

• Customers close the tab. If the browser disconnects mid-stream, the server usually stops the generator, but a model call already in progress may still finish and be billed. For long replies, that adds up; keep max_tokens sensible.
• Proxies that buffer. Some reverse proxies and hosting platforms collect the whole response before forwarding it, which silently turns your stream back into a 6-second wait. If streaming works locally but not in production, check the proxy’s buffering settings.
• Browsers on another domain. If the website at one domain calls your API at another, the browser blocks it unless the API allows it. That permission system is called CORS, and FastAPI has a CORSMiddleware to configure it. Allow only your own site’s domain.
• One user can spend your budget. Nothing here stops a script from calling /chat/stream ten thousand times. Per-user rate limiting and authentication are covered in Module 24.

Try it yourself

1. Put /triage, the three exception handlers and /chat/stream into one main.py and run uv run fastapi dev main.py.
2. Run load_test.py against /triage. Expected: all ten requests finish in roughly the time of one.
3. Temporarily change client = anthropic.AsyncAnthropic(...) to the sync client and the await client.messages.parse to a plain call, and rerun the load test. Expected: times climb 3, 6, 9… seconds. Change it back.
4. Set CLAUDE_MODEL=claude-does-not-exist in .env, restart, and call /triage. Expected: 502 with {"error": "The assistant failed to respond."} and a log line showing the provider’s 404.
5. Call /chat/stream with curl -N. Expected: text arrives a few words at a time.

Stuck? Hint

load_test.py needs httpx: run uv add httpx. If step 2 is still slow, check that every model call inside your endpoint has await in front of it and uses AsyncAnthropic. If step 5 shows everything at once, make sure you passed -N to curl.

Common mistakes

• A sync client inside async def. It freezes the event loop; every request waits for the one before. Use AsyncAnthropic and await, or a plain def endpoint.
• Creating a client per request. Slow, and it can exhaust connections under load. Create one at startup.
• Letting provider errors become bare 500s. Map them to 503 with Retry-After or 502, so callers know whether to retry.
• Expecting exception handlers to catch errors mid-stream. By then the status is sent. Handle them inside the generator.
• No input limits. Every endpoint that forwards text to a model needs a max_length.

Explaining it to Meera

Meera tried the website chat on her phone and watched the answer type itself out. “That’s much better. What did you change?”

“Three things. First, the system used to handle one question at a time, like one waiter standing at the kitchen door until each dish was ready. Now he takes other orders while he waits, so ten customers get answers in about three seconds instead of thirty.”

“Second?”

“When the AI company has a bad minute, we used to tell the WhatsApp system ‘we’re broken’, so it gave up for an hour. Now we say ‘busy, try again in thirty seconds’, and it does.”

“And the typing?”

“The AI writes its answer word by word anyway. Before, we waited for the last word and then showed everything. Now we pass each word on the moment it’s written. Same answer, same cost, but the first words appear in about a second.”

Recap

• async def only helps if everything slow inside is awaited; use AsyncAnthropic, or a plain def endpoint for sync code.
• Create one shared model client at startup and reuse its connections.
• Exception handlers translate provider errors into 503 (with Retry-After) or 502, with friendly messages; details go to logs.
• StreamingResponse plus messages.stream cuts time to first token from seconds to about one; errors mid-stream must be signalled inside the stream.
• Keep endpoints thin and put multi-step LLM workflows in service functions.

That evening Meera came back with one more request. “This is great for customers. But I want to try things myself, and so do the store managers. Not through curl, and not through the website. Can we have our own little chat page? By Friday?”

Go deeper

• FastAPI documentation (async, handling errors, custom responses, CORS): https://fastapi.tiangolo.com
• Anthropic docs (streaming): https://docs.anthropic.com
• HTTPX, used in the load test: https://www.python-httpx.org