Serving an LLM app with FastAPI: structure and async

What this guide covers

This guide is about the inside of a FastAPI application: how to arrange the code and how to use async correctly when most of a request's time is spent waiting for a model. The wider questions of modelling runs, state machines and durable workers are framework-neutral, and they are covered in building a backend API for an AI agent in Python.

The samples below are illustrative. FastAPI and model SDKs change over time, so check their current documentation before relying on exact signatures.

A project structure that stays manageable

The main idea is to keep HTTP concerns away from model concerns. Routes should translate between HTTP and your application. They should not build prompts or know which provider you are using. One reasonable layout looks like this:

app/
  main.py          # creates the app and registers routers
  config.py        # settings read from the environment
  dependencies.py  # shared objects supplied to routes
  routers/
    chat.py        # HTTP endpoints only
  schemas/
    chat.py        # request and response models
  services/
    chat.py        # application logic for one feature
  llm/
    client.py      # your small wrapper around the provider
    prompts.py     # prompt templates
tests/

Each layer gives you something specific. The wrapper in llm/ is the only place that knows a provider's details, so changing provider or adding a retry policy touches one module. The service layer can be tested without HTTP. The routers remain short enough to read at a glance. A small project does not need all of these folders on the first day, but the separation between routes, logic and provider code is worth having from the start.

Validate at the edge with schemas

Define request and response bodies as Pydantic models. FastAPI uses them to validate input and generate API documentation. For an LLM app, the request schema is also your first line of defence against oversized input, so set a maximum length on free text:

# Illustrative only.
from pydantic import BaseModel, Field

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

class ChatResponse(BaseModel):
    reply: str
    conversation_id: str

Keep these API schemas separate from any structure you ask the model to produce. Your public contract should not change because a prompt did, and model output needs its own validation before you trust it.

Share clients with lifespan and dependencies

Creating a new HTTP or model client for every request wastes connections and can hurt performance. Create long-lived clients once when the application starts, close them when it stops, and supply them to routes through dependencies. FastAPI's lifespan feature is designed for this kind of setup and teardown:

# Illustrative only. LLMClient is your own wrapper.
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI, Request

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.llm = LLMClient.from_settings()
    yield
    await app.state.llm.aclose()

app = FastAPI(lifespan=lifespan)

def get_llm(request: Request) -> LLMClient:
    return request.app.state.llm

@app.post("/chat", response_model=ChatResponse)
async def chat(body: ChatRequest, llm: LLMClient = Depends(get_llm)):
    reply = await llm.complete(body.message)
    return ChatResponse(reply=reply, conversation_id=body.conversation_id)

Dependencies also make testing simple. app.dependency_overrides lets a test replace get_llm with a fake client, so that route tests run without network access or cost.

Understanding async in an LLM service

An LLM call is a long wait on the network, and async suits that well. While one request waits for a response, the event loop can serve others. This only works if nothing blocks the loop, so it pays to know how FastAPI treats each kind of route:

• async def routes run on the event loop. Anything slow and blocking inside them stalls every other request handled by that worker.
• Plain def routes are run in a thread pool, so blocking calls will not stall the loop, although threads are a limited resource.

The most common mistake is calling a synchronous SDK or database driver from inside an async def route. It often seems fine under light testing, then the whole service becomes slow under concurrent use. Your options, in rough order of preference, are:

1. Use an async client, if your SDK or driver offers one.
2. Use a plain def route if everything it calls is synchronous.
3. Move the blocking call to a thread with asyncio.to_thread.
4. Send CPU-heavy work, such as local embedding or document parsing, to a separate process or worker. Threads do not generally help with CPU-bound Python, and an event loop certainly does not.

Running calls concurrently and limiting them

Async makes it easy to run independent steps together, such as retrieving from two sources at once with asyncio.gather. It also makes it easy to accept far more work than your provider or budget allows, because the event loop will happily start every request it receives. Set a deliberate limit, and fail clearly when it is reached:

# Illustrative only.
import asyncio
from fastapi import HTTPException

llm_slots = asyncio.Semaphore(20)  # example value; tune for your limits

async def complete_with_limits(llm, prompt):
    async with llm_slots:
        try:
            return await asyncio.wait_for(llm.complete(prompt), timeout=60)
        except asyncio.TimeoutError:
            raise HTTPException(status_code=504, detail="Model timed out")

Keep in mind that a semaphore belongs to one process. If you run several workers or instances, each has its own count, and a limit shared across all of them needs something external such as a rate limiter backed by a shared store. The same is true of any in-memory cache or conversation store: it disappears on restart and is invisible to other workers. Keep persistent state outside the process.

Set timeouts at every layer, too. An application timeout is of limited use if a proxy in front of it gives up earlier, so check that they agree with each other.

Streaming and client disconnects

StreamingResponse can wrap an async generator so that tokens are forwarded as they arrive:

# Illustrative only.
from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(body: ChatRequest, llm: LLMClient = Depends(get_llm)):
    async def events():
        async for token in llm.stream(body.message):
            yield f"data: {token}\n\n"
    return StreamingResponse(events(), media_type="text/event-stream")

Remember that the route has returned before the generator has finished. Resources created inside the route, or dependencies that clean up after the response, need checking so they are still available while streaming. If a user closes the tab, you also want to stop paying for tokens nobody will read, so confirm that disconnection actually stops the upstream call in your setup, and test it. Streaming LLM responses to a web UI covers event design, browsers and proxies.

When handling cancellation, avoid catching asyncio.CancelledError and carrying on. Clean up if you must, then let it propagate, or the task may continue working after it was asked to stop.

Background work: know the limits

FastAPI's BackgroundTasks runs a function after the response is sent, in the same process. It suits small follow-up jobs such as writing a log entry. It does not suit important or long-running work, because a restart can lose the task and nothing will retry it. Long agent runs generally belong in a durable queue with separate workers, and the route should simply record the job and return its ID.

Errors, logs and testing

• Map provider errors to your own. Catch them in the wrapper and raise your own exceptions. Use a FastAPI exception handler to turn them into consistent responses, so provider details do not leak to clients.
• Add a request ID. Middleware can attach one to each request and include it in the logs, which lets you connect an HTTP request to its model calls.
• Log duration and token usage per call. Redact sensitive content. This data is the starting point for controlling LLM cost and latency.
• Test with fakes. Override dependencies to simulate timeouts, malformed output and streams that fail partway. These cases are difficult to produce with a live model.
• Try concurrent requests. A simple load test against a slow fake model will quickly reveal a blocked event loop, because requests that should overlap will instead run one after another.

For help with the service, interface and deployment around a model, see full stack development for AI products.

Frequently asked questions

Should FastAPI routes for LLM calls be async?

If you have an async client, yes, since most of the time is spent waiting on the network. If everything the route calls is synchronous, use a plain def route, which FastAPI runs in a thread pool, instead of blocking the event loop.

Where should the model client be created in FastAPI?

Create long-lived clients once at startup using the lifespan feature, close them at shutdown and provide them to routes through dependencies. This reuses connections and lets tests substitute a fake.

Can FastAPI BackgroundTasks run long agent jobs?

It is a poor fit. Background tasks run in the same process, so a restart can lose them and there is no built-in retry. Use a durable queue with separate workers for long or important runs.

How do I limit concurrent LLM requests in FastAPI?

An asyncio semaphore works within one process. Across several workers or instances, each one has its own count, so use a shared rate limiter if you need a global limit.