FastAPI

FastAPI is our team's preferred framework for building production-grade APIs that serve AI workloads. It combines Python type hints, Pydantic validation, and native async support into a framework that feels natural for the AI engineering stack.

If you're building an API that calls LLMs, orchestrates agents, or serves embeddings, FastAPI is the right tool. This article covers the patterns we use in production — from request validation to streaming LLM responses to deployment on Azure Container Apps.

See also: Best Practices for our broader Python conventions, and LangChain for agent orchestration.

Why FastAPI

We evaluated Flask, Django REST Framework, and FastHTTP before settling on FastAPI. The decision came down to how well each framework supports the specific demands of AI engineering: high-latency I/O, structured data validation, and streaming responses.

What Makes FastAPI Fit

Native async support. Every layer — routing, middleware, response — is async-first. AI workloads are network-bound (LLM calls, vector DB queries), and async is the right model for them.
Automatic OpenAPI/Swagger docs. Every endpoint is documented at /docs and /redoc with no extra configuration. Your API is self-documenting.
Pydantic validation built-in. Request bodies, query parameters, and path parameters are validated at the framework level. Invalid data never reaches your business logic.
Type hints everywhere. FastAPI uses Python type annotations for routing, validation, serialization, and documentation. One annotation does four jobs.

Comparison Table

Feature	FastAPI	Flask	Django REST Framework
Async support	Native	Limited (3.x beta)	Limited
Request validation	Pydantic (automatic)	Manual (Werkzeug)	Serializers (verbose)
API docs	Auto-generated (OpenAPI)	Manual (Stoplight, etc.)	Manual (DRF Spectacular)
Type safety	Built-in (type hints)	None	None
Performance	High (Starlette/Uvicorn)	Medium (Werkzeug)	Medium (WSGI)
Streaming responses	First-class	Add-on	Add-on
Dependency injection	Built-in	Third-party	Built-in (different model)
Learning curve	Low (if you know Python typing)	Low	High
AI workload fit	Excellent	Fair	Poor

Why not Flask?

Flask is great for simple scripts and prototypes. But for production AI services, you end up adding extensions for everything: validation, async, API docs, CORS. FastAPI ships with all of this built-in. Use Flask only when you need a minimal script with no structure.

Your First Endpoint

Installation

# Add FastAPI and the ASGI server
uv add fastapi uvicorn

# Sync environment
uv sync

Minimal App

Create main.py:

from fastapi import FastAPI

app = FastAPI(title="AI Service", version="0.1.0")


@app.get("/")
async def root() -> dict[str, str]:
    return {"message": "AI service is running"}


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "healthy"}

Running the Server

uv run uvicorn main:app --reload

The --reload flag enables auto-reload during development. Omit it in production.

Open your browser:

http://localhost:8000/ — Root endpoint
http://localhost:8000/health — Health check
http://localhost:8000/docs — Auto-generated Swagger UI
http://localhost:8000/redoc — Alternative ReDoc documentation

The main:app syntax

uvicorn main:app means "load the app object from the main module." If your app lives in src/my_service/app.py, the command becomes uvicorn my_service.app:app.

Request Validation with Pydantic

Pydantic is FastAPI's built-in validation engine. You define models using BaseModel, and FastAPI handles parsing, type coercion, and error responses automatically.

Defining Request Models

from pydantic import BaseModel, Field


class ChatRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=4096, description="User prompt")
    model: str = Field(default="gpt-4o", description="LLM model to use")
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(default=1024, ge=1, le=8192)

Using Models in Endpoints

from fastapi import FastAPI

app = FastAPI()


@app.post("/chat")
async def chat(request: ChatRequest) -> dict[str, str]:
    # FastAPI already validated the request body.
    # If validation fails, it returns 422 automatically.
    return {
        "model": request.model,
        "response": f"Processed prompt with {request.max_tokens} max tokens",
    }

Path, Query, and Body Parameters

FastAPI distinguishes parameter sources by type hints:

from fastapi import FastAPI, Query, Path

app = FastAPI()


@app.get("/models/{model_id}")
async def get_model(
    model_id: str = Path(min_length=1, description="Model identifier"),
    include_details: bool = Query(default=False, description="Include full metadata"),
) -> dict:
    return {"model_id": model_id, "details": include_details}

Automatic Validation Errors

When a request fails validation, FastAPI returns a structured 422 Unprocessable Entity response:

curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"prompt": "", "temperature": 5.0}'

{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "prompt"],
      "msg": "String should have at least 1 character",
      "input": ""
    },
    {
      "type": "less_than_equal",
      "loc": ["body", "temperature"],
      "msg": "Input should be less than or equal to 2",
      "input": 5.0
    }
  ]
}

Do and Don't

Do
Don't

from pydantic import BaseModel, Field


# Define models with constraints
class EmbeddingRequest(BaseModel):
    text: str = Field(min_length=1, max_length=8000)
    dimensions: int = Field(default=1536, ge=256, le=4096)


# Let FastAPI handle validation
@app.post("/embed")
async def embed(request: EmbeddingRequest) -> dict:
    return {"dimensions": request.dimensions, "status": "queued"}

# Bad: Manual validation in endpoint logic
@app.post("/embed")
async def embed(text: str, dimensions: int = 1536) -> dict:
    if not text:
        return {"error": "text is required"}, 400
    if len(text) > 8000:
        return {"error": "text too long"}, 400
    if dimensions < 256 or dimensions > 4096:
        return {"error": "invalid dimensions"}, 400
    # Business logic buried in validation...
    return {"dimensions": dimensions, "status": "queued"}

Validation at the boundary

Keep validation in Pydantic models, not in endpoint functions. Your endpoint logic should assume data is already valid. This keeps business logic clean and error handling centralized.

Response Models

Response models control what your API sends back. They filter, transform, and document your output.

Basic Response Model

from pydantic import BaseModel


class ChatResponse(BaseModel):
    response: str
    model: str
    usage: dict[str, int]


@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest) -> ChatResponse:
    # Only fields in ChatResponse are returned.
    # Extra fields in the return value are silently dropped.
    return ChatResponse(
        response="The answer is 42.",
        model=request.model,
        usage={"prompt_tokens": 10, "completion_tokens": 5},
    )

Filtering Response Fields

Use response_model to exclude sensitive or internal fields:

from pydantic import BaseModel


class AgentInternal(BaseModel):
    prompt_template: str
    tools: list[str]
    api_key: str  # Internal — should not leak


class AgentPublic(BaseModel):
    name: str
    version: str
    capabilities: list[str]


# Return only the public shape
@app.get("/agent", response_model=AgentPublic)
async def get_agent() -> AgentPublic:
    return AgentPublic(
        name="Research Agent",
        version="1.0.0",
        capabilities=["web_search", "code_execution"],
    )

Nested Models

from pydantic import BaseModel


class ToolResult(BaseModel):
    name: str
    output: str
    latency_ms: float


class AgentStep(BaseModel):
    step_number: int
    thought: str
    tool_used: str | None = None
    result: ToolResult | None = None


class AgentTrace(BaseModel):
    query: str
    steps: list[AgentStep]
    final_answer: str
    total_latency_ms: float


@app.get("/trace/{trace_id}", response_model=AgentTrace)
async def get_trace(trace_id: str) -> AgentTrace:
    return AgentTrace(
        query="What is the capital of France?",
        steps=[
            AgentStep(
                step_number=1,
                thought="I need to search for this.",
                tool_used="web_search",
                result=ToolResult(name="web_search", output="Paris", latency_ms=230),
            )
        ],
        final_answer="The capital of France is Paris.",
        total_latency_ms=340,
    )

Custom Status Codes and Content Types

from fastapi import Response


@app.post("/chat/stream")
async def chat_stream(
    request: ChatRequest,
    response: Response,
) -> Response:
    # Return a custom response with specific headers
    response.headers["X-Request-Id"] = "req-123"
    return Response(
        content="Streaming response content",
        media_type="text/event-stream",
        status_code=200,
    )

Dependency Injection

FastAPI's dependency injection system replaces traditional middleware for many use cases. It is declarative, testable, and composable.

Basic Dependency

from fastapi import Depends, FastAPI


def get_api_key() -> str:
    """Return the API key from environment."""
    import os
    return os.environ["OPENAI_API_KEY"]


app = FastAPI()


@app.get("/models")
async def list_models(api_key: str = Depends(get_api_key)) -> dict:
    # api_key is injected automatically
    return {"authenticated": True, "key_prefix": api_key[:5]}

Database Session Management

from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI
import httpx


async def get_http_client() -> AsyncIterator[httpx.AsyncClient]:
    """Provide an HTTP client with connection pooling."""
    async with httpx.AsyncClient(timeout=30.0) as client:
        yield client


app = FastAPI()


@app.get("/search")
async def search(
    query: str,
    client: httpx.AsyncClient = Depends(get_http_client),
) -> dict:
    # The client is shared across requests that use this dependency.
    # FastAPI handles lifecycle (open/close).
    response = await client.get(
        "https://api.search.com",
        params={"q": query},
    )
    return response.json()

Auth Middleware via Dependencies

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


class User(BaseModel):
    id: str
    role: str


def get_current_user(authorization: str | None = Header(None)) -> User:
    """Extract and validate the current user from the Authorization header."""
    if not authorization:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing authorization header",
        )
    if not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authorization scheme",
        )
    token = authorization.removeprefix("Bearer ")
    # In production, verify the token with your auth provider.
    return User(id="user-123", role="admin")


@app.get("/protected")
async def protected_route(user: User = Depends(get_current_user)) -> dict:
    return {"user_id": user.id, "role": user.role}

Lazy Loading

Dependencies are resolved lazily — only when the endpoint is called:

from fastapi import Depends, FastAPI


def get_llm_client():
    """Expensive initialization — only runs when needed."""
    from langchain_openai import ChatOpenAI
    print("Initializing LLM client...")  # Only prints on first request
    return ChatOpenAI(model="gpt-4o")


app = FastAPI()


@app.post("/chat")
async def chat(
    request: ChatRequest,
    llm = Depends(get_llm_client),
) -> dict:
    result = await llm.ainvoke(request.prompt)
    return {"response": result.content}

Dependency vs. Middleware

Concern	Use Dependency	Use Middleware
Per-endpoint logic (auth, DB session)	Yes	No
Cross-cutting (logging, CORS, timing)	No	Yes
Testable in isolation	Yes	Harder
Conditional on route	Yes	No
Global request/response transform	No	Yes

Rule of thumb

Use dependencies for per-request business logic (auth, database sessions, LLM clients). Use middleware for global cross-cutting concerns (logging, CORS, request IDs).

Async Patterns for AI Workloads

AI engineering is inherently async. LLM calls, vector database queries, and tool executions are all network-bound operations. FastAPI's async support is critical for throughput.

`async def` vs `def`

FastAPI handles both, but the semantics differ:

from fastapi import FastAPI

app = FastAPI()


# Runs in the event loop — use for I/O-bound work
@app.get("/async-endpoint")
async def async_endpoint() -> dict:
    # Non-blocking — other requests can be handled while waiting
    result = await some_async_llm_call("prompt")
    return {"result": result}


# Runs in a thread pool — use for CPU-bound or sync libraries
@app.get("/sync-endpoint")
def sync_endpoint() -> dict:
    # Runs in a thread pool executor (does not block the event loop)
    result = sync_heavy_computation()
    return {"result": result}

Pattern	When to Use
`async def`	Calling async libraries (httpx, langchain `ainvoke`, async DB)
`def`	Calling sync libraries, CPU-bound work, or code you cannot make async

Blocking the event loop

Never use blocking calls (time.sleep, synchronous requests.get, synchronous file I/O) inside an async def handler. This blocks the entire event loop and kills throughput. Use def (thread pool) or async def (async I/O) appropriately.

Streaming Responses for LLM Output

Streaming is essential for LLM APIs. Users see tokens appear in real time instead of waiting for the full response:

from collections.abc import AsyncIterator
from fastapi import FastAPI
from fastapi.responses import StreamingResponse


async def stream_llm_response(prompt: str) -> AsyncIterator[str]:
    """Stream tokens from an LLM call."""
    # In production, replace with your LLM client's streaming API.
    tokens = ["The", " capital", " of", " France", " is", " Paris", "."]
    for token in tokens:
        yield token
        await asyncio.sleep(0.1)  # Simulate network latency


app = FastAPI()


@app.post("/chat/stream")
async def chat_stream(request: ChatRequest) -> StreamingResponse:
    return StreamingResponse(
        content=stream_llm_response(request.prompt),
        media_type="text/event-stream",
    )

Server-Sent Events (SSE)

For structured streaming with metadata (token counts, latency, etc.), use SSE:

import json
import asyncio
from collections.abc import AsyncIterator
from fastapi import FastAPI
from fastapi.responses import StreamingResponse


async def sse_llm_stream(prompt: str) -> AsyncIterator[str]:
    """Stream LLM response as Server-Sent Events."""
    tokens = ["Hello", " from", " the", " AI", " service", "!"]
    token_count = 0

    for token in tokens:
        token_count += 1
        data = json.dumps({
            "token": token,
            "index": token_count,
        })
        yield f"data: {data}\n\n"
        await asyncio.sleep(0.1)

    # Send a final event with usage stats
    final = json.dumps({
        "type": "done",
        "usage": {"total_tokens": token_count},
    })
    yield f"data: {final}\n\n"


@app.post("/chat/sse")
async def chat_sse(request: ChatRequest) -> StreamingResponse:
    return StreamingResponse(
        content=sse_llm_stream(request.prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
        },
    )

Background Tasks

For operations that should not block the response:

from fastapi import BackgroundTasks, FastAPI


async def log_interaction(user_id: str, prompt: str, response: str) -> None:
    """Log the interaction to a database or analytics service."""
    # In production, write to PostgreSQL, send to Langfuse, etc.
    await asyncio.sleep(0.1)  # Simulate DB write


app = FastAPI()


@app.post("/chat")
async def chat_with_logging(
    request: ChatRequest,
    background_tasks: BackgroundTasks,
) -> dict:
    # Simulate LLM call
    response = "AI response here"

    # Schedule logging — does not block the response
    background_tasks.add_task(
        log_interaction,
        user_id="user-123",
        prompt=request.prompt,
        response=response,
    )

    return {"response": response}

Concurrent Requests

FastAPI handles concurrent requests out of the box:

import asyncio
from fastapi import FastAPI


async def call_llm(prompt: str) -> str:
    """Simulate an LLM call with latency."""
    await asyncio.sleep(2.0)  # 2-second LLM call
    return f"Response to: {prompt}"


app = FastAPI()


@app.post("/batch")
async def batch_prompts(prompts: list[str]) -> list[str]:
    """Process multiple prompts concurrently."""
    # All LLM calls run in parallel
    results = await asyncio.gather(
        *[call_llm(prompt) for prompt in prompts]
    )
    return list(results)

Error Handling

Production APIs need consistent, structured error responses.

HTTPException for Expected Errors

from fastapi import FastAPI, HTTPException, status


app = FastAPI()


@app.get("/models/{model_id}")
async def get_model(model_id: str) -> dict:
    available_models = ["gpt-4o", "gpt-4o-mini", "o3-mini"]

    if model_id not in available_models:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Model '{model_id}' not found. Available: {available_models}",
        )

    return {"model_id": model_id, "status": "available"}

Custom Exception Handlers

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class AIServiceError(Exception):
    """Base exception for AI service errors."""
    def __init__(self, message: str, code: str = "ai_service_error") -> None:
        self.message = message
        self.code = code


class LLMTimeoutError(AIServiceError):
    """Raised when an LLM call times out."""
    def __init__(self, message: str = "LLM call timed out") -> None:
        super().__init__(message, code="llm_timeout")


class LLMRateLimitError(AIServiceError):
    """Raised when rate-limited by the provider."""
    def __init__(self, message: str = "Rate limited by LLM provider") -> None:
        super().__init__(message, code="llm_rate_limit")


app = FastAPI()


@app.exception_handler(AIServiceError)
async def ai_service_error_handler(
    request: Request,
    exc: AIServiceError,
) -> JSONResponse:
    status_map = {
        "llm_timeout": status.HTTP_504_GATEWAY_TIMEOUT,
        "llm_rate_limit": status.HTTP_429_TOO_MANY_REQUESTS,
    }
    return JSONResponse(
        status_code=status_map.get(exc.code, status.HTTP_500_INTERNAL_SERVER_ERROR),
        content={
            "error": {
                "code": exc.code,
                "message": exc.message,
            },
        },
    )

Global Error Handler

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


app = FastAPI()


@app.exception_handler(Exception)
async def global_exception_handler(
    request: Request,
    exc: Exception,
) -> JSONResponse:
    """Catch-all handler for unhandled exceptions."""
    # In production, log the full traceback here.
    import logging
    logging.exception("Unhandled exception")

    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={
            "error": {
                "code": "internal_error",
                "message": "An unexpected error occurred",
            },
        },
    )

Structured Error Responses with Pydantic

from pydantic import BaseModel


class ErrorDetail(BaseModel):
    code: str
    message: str
    field: str | None = None


class ErrorResponse(BaseModel):
    error: ErrorDetail


# Use in exception handlers for consistent shape
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content=ErrorResponse(
            error=ErrorDetail(
                code="validation_error",
                message=str(exc),
            )
        ).model_dump(),
    )

Middleware

Middleware runs on every request, before dependencies and route handlers. Use it for cross-cutting concerns.

Request Logging

import time
import logging
from fastapi import FastAPI, Request

logger = logging.getLogger(__name__)


@app.middleware("http")
async def log_requests(request: Request, call_next) -> Response:
    start_time = time.perf_counter()

    response = await call_next(request)

    elapsed = time.perf_counter() - start_time
    logger.info(
        "%s %s %s %.3fms",
        request.method,
        request.url.path,
        response.status_code,
        elapsed * 1000,
    )
    return response

CORS Configuration

from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.app"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Request ID Tracking

Critical for correlating requests with Langfuse traces:

import uuid
from fastapi import FastAPI, Request, Response


@app.middleware("http")
async def add_request_id(request: Request, call_next) -> Response:
    # Generate a unique request ID for trace correlation
    request_id = request.headers.get("X-Request-Id", str(uuid.uuid4()))
    request.state.request_id = request_id

    response = await call_next(request)
    response.headers["X-Request-Id"] = request_id

    return response


# Access the request ID in endpoints
@app.get("/chat")
async def chat(request: Request) -> dict:
    return {"request_id": request.state.request_id}

Langfuse Correlation

When integrating with Langfuse, pass the request ID as the trace_id parameter. This links API requests to LLM traces in the Langfuse UI:

from langfuse import Langfuse

langfuse = Langfuse()

@app.post("/chat")
async def chat(request: ChatRequest, req: Request) -> dict:
    trace = langfuse.trace(
        name="chat-request",
        id=req.state.request_id,  # Correlate with API request
        input={"prompt": request.prompt},
    )
    # ... call LLM ...
    return {"response": "AI response"}

Rate Limiting

import time
from collections import defaultdict
from fastapi import FastAPI, Request, HTTPException, status


class RateLimiter:
    def __init__(self, max_requests: int = 60, window_seconds: int = 60) -> None:
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests: dict[str, list[float]] = defaultdict(list)

    def is_allowed(self, key: str) -> bool:
        now = time.time()
        # Clean old entries
        self.requests[key] = [
            t for t in self.requests[key]
            if now - t < self.window_seconds
        ]
        if len(self.requests[key]) >= self.max_requests:
            return False
        self.requests[key].append(now)
        return True


rate_limiter = RateLimiter(max_requests=60, window_seconds=60)


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next) -> Response:
    client_ip = request.client.host
    if not rate_limiter.is_allowed(client_ip):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="Rate limit exceeded. Try again later.",
        )
    return await call_next(request)

Testing

Test FastAPI apps with pytest. Use TestClient for sync tests and AsyncClient for async tests.

TestClient for Sync Tests

# tests/test_endpoints.py
import pytest
from fastapi.testclient import TestClient
from main import app


@pytest.fixture
def client() -> TestClient:
    return TestClient(app)


def test_root(client: TestClient) -> None:
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "AI service is running"}


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


def test_chat_validation(client: TestClient) -> None:
    response = client.post(
        "/chat",
        json={"prompt": "", "temperature": 5.0},
    )
    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any("prompt" in str(err) for err in errors)
    assert any("temperature" in str(err) for err in errors)

AsyncClient for Async Tests

import pytest
from httpx import AsyncClient, ASGITransport
from main import app


@pytest.fixture
async def async_client() -> AsyncClient:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as client:
        yield client


@pytest.mark.asyncio
async def test_chat_stream(async_client: AsyncClient) -> None:
    response = await async_client.post(
        "/chat/stream",
        json={"prompt": "Hello", "model": "gpt-4o-mini"},
    )
    assert response.status_code == 200
    assert response.headers["content-type"] == "text/event-stream; charset=utf-8"

Dependency Overriding

Override dependencies to inject test doubles:

from unittest.mock import AsyncMock
from fastapi.testclient import TestClient
from main import app, get_llm_client


def test_chat_with_mock_llm() -> None:
    # Override the LLM client dependency
    mock_llm = AsyncMock()
    mock_llm.ainvoke.return_value.content = "Mocked response"
    app.dependency_overrides[get_llm_client] = lambda: mock_llm

    try:
        with TestClient(app) as client:
            response = client.post(
                "/chat",
                json={"prompt": "Test prompt", "model": "gpt-4o"},
            )
            assert response.status_code == 200
            assert response.json()["response"] == "Mocked response"
    finally:
        # Always clear overrides to avoid test pollution
        app.dependency_overrides.clear()

Fixture Patterns

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from main import app


@pytest.fixture
def client() -> TestClient:
    """Return a TestClient for the FastAPI app."""
    return TestClient(app)


@pytest.fixture
def sample_chat_request() -> dict:
    """Return a valid chat request payload."""
    return {
        "prompt": "What is the capital of France?",
        "model": "gpt-4o-mini",
        "temperature": 0.7,
        "max_tokens": 256,
    }


@pytest.fixture
def invalid_chat_request() -> dict:
    """Return an invalid chat request payload."""
    return {
        "prompt": "",
        "temperature": 5.0,
    }


# tests/test_chat.py
def test_chat_success(client: TestClient, sample_chat_request: dict) -> None:
    response = client.post("/chat", json=sample_chat_request)
    assert response.status_code == 200


def test_chat_validation_error(client: TestClient, invalid_chat_request: dict) -> None:
    response = client.post("/chat", json=invalid_chat_request)
    assert response.status_code == 422

Always clear dependency overrides

Use try/finally or a fixture with yield to ensure app.dependency_overrides.clear() runs even if the test fails. Stale overrides cause mysterious test failures.

Production Deployment

Uvicorn Workers

For development, uvicorn main:app --reload is sufficient. For production, configure workers explicitly:

# Single worker (good for container environments)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1

# Multiple workers (good for bare metal or VMs)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

Gunicorn + Uvicorn for Multi-Worker

Gunicorn manages worker processes; Uvicorn handles async within each worker:

# Install gunicorn
uv add gunicorn

# Run with gunicorn using uvicorn workers
gunicorn main:app \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --bind 0.0.0.0:8000 \
  --timeout 120 \
  --keep-alive 5

The --timeout 120 is important for AI workloads — LLM calls can take longer than Gunicorn's default 30-second timeout.

Health Check Endpoint

Container orchestration platforms need a health check endpoint:

from fastapi import FastAPI

app = FastAPI()


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "healthy"}


@app.get("/ready")
async def readiness() -> dict[str, str]:
    """Check if the service is ready to accept traffic."""
    # In production, check downstream dependencies:
    # - Database connectivity
    # - LLM provider availability
    # - Cache connectivity
    return {"status": "ready"}

Configure the health check in your Azure Container Apps deployment:

# azure-container-apps deployment config
template:
  containers:
    - name: ai-service
      image: your-registry/ai-service:latest
      resources:
        cpu: 1.0
        memory: 2Gi
      probes:
        - type: health
          path: /health
          periodSeconds: 10
          initialDelaySeconds: 5

Structured JSON Logging

import logging
import sys
import json


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }
        if hasattr(record, "request_id"):
            log_entry["request_id"] = record.request_id
        if record.exc_info:
            log_entry["exception"] = self.formatException(record.exc_info)
        return json.dumps(log_entry)


# Configure root logger
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logging.basicConfig(handlers=[handler], level=logging.INFO)

Container logging

In containerized environments, structured JSON logs integrate with observability tools (Azure Monitor, Datadog, Langfuse). Plain text logs require parsing; JSON logs are machine-readable.

Dockerfile for AI Services

FROM python:3.12-slim

WORKDIR /app

# Install uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv

# Copy dependency files
COPY pyproject.toml uv.lock ./

# Install dependencies (cached layer)
RUN uv sync --frozen --no-cache

# Copy application code
COPY src/ src/

# Run with uvicorn
CMD ["uv", "run", "uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

Key Takeaways

Summary

FastAPI is async-first — use async def for I/O-bound AI workloads (LLM calls, vector DB queries) and def for CPU-bound or sync library calls.
Pydantic models are your validation layer — define request and response models with BaseModel and Field constraints. Let FastAPI handle 422 errors.
Dependency injection replaces middleware for per-request logic (auth, DB sessions, LLM clients). Use middleware for global concerns (logging, CORS, request IDs).
StreamingResponse for LLM output — stream tokens to users instead of waiting for full responses. Use SSE for structured streaming with metadata.
BackgroundTasks for non-blocking work — log interactions, update analytics, or trigger downstream jobs without delaying the response.
Test with TestClient and override dependencies — mock LLM clients and database sessions to keep tests fast and deterministic.
Production deployment uses Gunicorn + Uvicorn — configure timeouts for LLM latency and health checks for container orchestration.
Request ID middleware enables trace correlation — link API requests to Langfuse traces for full observability.

Why FastAPI​

What Makes FastAPI Fit​

Comparison Table​

Your First Endpoint​

Installation​

Minimal App​

Running the Server​

Request Validation with Pydantic​

Defining Request Models​

Using Models in Endpoints​

Path, Query, and Body Parameters​

Automatic Validation Errors​

Do and Don't​

Response Models​

Basic Response Model​

Filtering Response Fields​

Nested Models​

Custom Status Codes and Content Types​

Dependency Injection​

Basic Dependency​

Database Session Management​

Auth Middleware via Dependencies​

Lazy Loading​

Dependency vs. Middleware​

Async Patterns for AI Workloads​

async def vs def​

Streaming Responses for LLM Output​

Server-Sent Events (SSE)​

Background Tasks​

Concurrent Requests​

Error Handling​

HTTPException for Expected Errors​

Custom Exception Handlers​

Global Error Handler​

Structured Error Responses with Pydantic​

Middleware​

Request Logging​

CORS Configuration​

Request ID Tracking​

Rate Limiting​

Testing​

TestClient for Sync Tests​

AsyncClient for Async Tests​

Dependency Overriding​

Fixture Patterns​

Production Deployment​

Uvicorn Workers​

Gunicorn + Uvicorn for Multi-Worker​

Health Check Endpoint​

Structured JSON Logging​

Dockerfile for AI Services​

Key Takeaways​

Further Reading​