Building GenAI Applications with FastAPI — II

DhanushKumar

7 min read

·

Jul 9, 2026

--

Listen

Share

Following the previous blog : https://medium.com/@danushidk507/building-genai-applications-with-fastapi-i-2174827f30bd?sharedUserId=danushidk507 ,

You’ve got a working LLM prototype. Now what? This guide walks you through everything you need to take it to production — from streaming responses and RAG pipelines to auth, rate limiting, WebSockets, background jobs, and deployment. All through the lens of building real GenAI applications.

Why FastAPI for GenAI?

Most GenAI tutorials show you how to call an LLM. Almost none show you how to build the API layer that makes it production-worthy.

FastAPI is the default choice for Python AI backends for very good reasons:

• Async-native — LLM calls are I/O bound. async/await lets you serve hundreds of concurrent requests without thread overhead
• Streaming support — Token-by-token streaming is a first-class citizen via StreamingResponse
• Pydantic validation — Prompt inputs, model configs, and structured outputs are validated automatically
• Auto-generated docs — Your Swagger UI becomes a live playground for your AI endpoints
• Dependency injection — LLM clients, vector DB connections, and auth are shared cleanly across routes

Testing Your GenAI API

Testing AI endpoints has unique challenges: non-deterministic outputs, expensive external API calls, and slow embedding pipelines. The solution: mock aggressively, test structure not content.

from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, patch, MagicMock
import pytest

client = TestClient(app)
# Mock OpenAI response
def mock_openai_response(content: str = "This is a test reply"):
    mock = MagicMock()
    mock.choices[0].message.content = content
    mock.usage.total_tokens = 42
    mock.usage.prompt_tokens = 10
    mock.usage.completion_tokens = 32
    return mock
# Override LLM client in tests
@pytest.fixture
def mock_llm():
    with patch("app.dependencies.llm.get_openai_client") as mock:
        client = AsyncMock()
        client.chat.completions.create = AsyncMock(
            return_value=mock_openai_response()
        )
        mock.return_value = client
        yield client
# Override database
@pytest.fixture
def mock_db():
    app.dependency_overrides[get_db] = lambda: MagicMock()
    yield app.dependency_overrides.clear()
# Tests
def test_chat_endpoint_structure(mock_llm, mock_db):
    response = client.post(
        "/chat",
        json={
            "messages": [{"role": "user", "content": "Hello"}],
            "model": "gpt-4o",
        },
        headers={"X-API-Key": "test-key"},
    )
    assert response.status_code == 200
    data = response.json()
    assert "reply" in data
    assert "tokens_used" in data
    assert "model" in data
def test_chat_validation_rejects_bad_role():
    response = client.post(
        "/chat",
        json={"messages": [{"role": "invalid_role", "content": "Hi"}]},
    )
    assert response.status_code == 422
def test_streaming_returns_event_stream(mock_llm):
    response = client.post(
        "/chat/stream",
        json={"messages": [{"role": "user", "content": "Test"}], "stream": True},
        headers={"X-API-Key": "test-key"},
    )
    assert response.status_code == 200
    assert "text/event-stream" in response.headers["content-type"]
def test_rate_limit_returns_429():
    with patch("app.routers.chat.check_token_budget") as mock:
        mock.side_effect = HTTPException(429, "Rate limit exceeded")
        response = client.post("/chat", json={"messages": [{"role": "user", "content": "Hi"}]})
        assert response.status_code == 429
def test_rag_document_ingestion(mock_db):
    response = client.post(
        "/rag/documents",
        json={
            "content": "FastAPI is a modern web framework for Python.",
            "title": "FastAPI Overview",
            "collection": "test",
        },
        headers={"X-API-Key": "test-key"},
    )
    assert response.status_code == 202
    assert "doc_id" in response.json()
    assert response.json()["status"] == "processing"
# Async tests
@pytest.mark.asyncio
async def test_websocket_chat():
    from httpx import AsyncClient
    async with AsyncClient(app=app, base_url="http://test") as ac:
        with client.websocket_connect("/ws/chat/test-session") as ws:
            # Receive history on connect
            msg = ws.receive_json()
            assert msg["type"] == "history"
            # Send a message
            ws.send_json({"type": "message", "content": "Hello"})
            thinking = ws.receive_json()
            assert thinking["type"] == "thinking"

Advanced Features for GenAI APIs

The Request Object — Reading Raw Prompts

Request allows you to access the underlying HTTP request object directly within your path operation functions or middlewares. This object is a direct subclass of the Starlette Request class

from fastapi import Request

@app.post("/debug/request")
async def debug_request(request: Request):
    body = await request.body()
    return {
        "method": request.method,
        "client_ip": request.client.host,
        "content_type": request.headers.get("content-type"),
        "body_size_bytes": len(body),
        "trace_id": request.headers.get("x-trace-id"),
    }

status Module — Semantic Status Codes for AI Ops

In FastAPI, importing from fastapi import status provides you with explicit, human-readable names for HTTP response status codes. This object is a direct import of Starlette’s status module, eliminating the need to memorize arbitrary numeric HTTP codes like 201 or 404.

from fastapi import status

@app.post("/documents", status_code=status.HTTP_202_ACCEPTED)  # async ingestion
async def ingest():
    return {"status": "queued"}
@app.delete("/sessions/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def clear_session(id: str):
    return None
@app.post("/models/{name}/load", status_code=status.HTTP_201_CREATED)
async def load_model(name: str):
    return {"loaded": name}

StaticFiles — Serving Your AI Chat Frontend

from fastapi.staticfiles import StaticFiles

# Serve bundled React/Next.js chat UI
app.mount("/", StaticFiles(directory="frontend/dist", html=True), name="frontend")
# Note: mount API routes BEFORE the static mount, or they'll be shadowed

Jinja2 — Server-Side AI Report Rendering

from fastapi.templating import Jinja2Templates

templates = Jinja2Templates(directory="templates")
@app.get("/reports/{report_id}", response_class=HTMLResponse)
async def render_report(request: Request, report_id: str):
    report = await report_store.get(report_id)
    return templates.TemplateResponse(
        "report.html",
        {
            "request": request,
            "title": report.title,
            "summary": report.ai_summary,
            "sections": report.sections,
            "generated_at": report.created_at,
        },
    )

url_path_for — Building Callback & Webhook URLs

@app.get("/agents/runs/{run_id}", name="get_agent_run")
async def get_run(run_id: str):
    return await agent_store.get(run_id)

@app.post("/agents/run")
async def start_run(request: AgentRunRequest, http_request: Request, background_tasks: BackgroundTasks):
    run_id = f"run_{uuid.uuid4().hex[:16]}"
    # Build the polling URL dynamically
    poll_url = str(http_request.url_for("get_agent_run", run_id=run_id))
    background_tasks.add_task(execute_agent, run_id, request)
    return {"run_id": run_id, "poll_url": poll_url}
    # Returns: { "poll_url": "http://api.example.com/agents/runs/run_abc123" }

OpenAPI Customization — Your AI API as a Product

A well-documented AI API is a product in itself. Customize your OpenAPI schema to make it developer-friendly.

from fastapi.openapi.utils import get_openapi

tags_metadata = [
    {
        "name": "Chat Completions",
        "description": "Send messages and receive AI-generated responses. Supports streaming via SSE.",
    },
    {
        "name": "RAG & Knowledge Base",
        "description": "Retrieve-Augmented Generation. Ingest documents and query them with natural language.",
        "externalDocs": {"description": "RAG Guide", "url": "https://docs.yourapi.com/rag"},
    },
    {
        "name": "Embeddings",
        "description": "Convert text to vector embeddings for semantic search and similarity.",
    },
    {
        "name": "Agents",
        "description": "Autonomous AI agents that use tools to complete multi-step tasks.",
    },
    {
        "name": "Admin",
        "description": "Usage analytics, model management, and billing. Requires admin role.",
    },
]
def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    schema = get_openapi(
        title="GenAI API",
        version="1.0.0",
        description="""
## The AI Backend API
Supports completions, streaming, RAG, embeddings, and agent orchestration.
### Authentication
All endpoints require either:
- **API Key** in `X-API-Key` header
- **Bearer Token** in `Authorization` header
### Rate Limits
| Tier | Tokens/min | Requests/min |
|---|---|---|
| Free | 10,000 | 20 |
| Pro | 100,000 | 200 |
| Enterprise | Unlimited | Custom |
        """,
        routes=app.routes,
        tags=tags_metadata,
    )
    # Add security schemes
    schema["components"]["securitySchemes"] = {
        "ApiKeyAuth": {"type": "apiKey", "in": "header", "name": "X-API-Key"},
        "BearerAuth": {"type": "http", "scheme": "bearer"},
    }
    schema["security"] = [{"ApiKeyAuth": []}, {"BearerAuth": []}]
    app.openapi_schema = schema
    return schema
app.openapi = custom_openapi

Document multiple response codes for AI-specific failures:

@app.post(
    "/chat",
    responses={
        200: {"description": "Completion successful", "model": ChatResponse},
        400: {"description": "Context length exceeded or content filtered"},
        401: {"description": "Invalid or missing API key"},
        402: {"description": "Token quota exceeded — upgrade required"},
        429: {"description": "Rate limit exceeded", "headers": {"Retry-After": {"schema": {"type": "integer"}}}},
        503: {"description": "LLM provider temporarily unavailable"},
    },
)
async def chat(request: ChatRequest):
    ...

Deployment — Taking Your GenAI API to Production

Environment Configuration .It allows you to automatically load and validate environment variables or .env files into strongly typed Python objects

from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # Server
    env: str = "dev"
    debug: bool = False
    # LLM Providers
    openai_api_key: str
    anthropic_api_key: Optional[str] = None
    google_api_key: Optional[str] = None
    default_model: str = "gpt-4o"
    # Databases
    database_url: str
    redis_url: str = "redis://localhost:6379"
    chroma_host: str = "localhost"
    chroma_port: int = 8001
    # Auth
    jwt_secret: str
    api_key_salt: str
    # Limits
    max_tokens_per_request: int = 4096
    rate_limit_tokens_per_minute: int = 100000
    max_upload_size_mb: int = 10
    class Config:
        env_file = ".env"
settings = Settings()

Dockerfile

FROM python:3.11-slim

WORKDIR /app
# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application
COPY . .
# Non-root user for security
RUN useradd -m appuser && chown -R appuser /app
USER appuser
EXPOSE 8000
# Production: multiple workers, no reload
CMD ["gunicorn", "main:app", \
     "-w", "4", \
     "-k", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000", \
     "--timeout", "120", \
     "--keep-alive", "5"]

Docker Compose (Full GenAI Stack)

version: "3.9"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - ENV=production
      - DATABASE_URL=postgresql+asyncpg://user:pass@postgres/genai
      - REDIS_URL=redis://redis:6379
      - CHROMA_HOST=chroma
    depends_on:
      - postgres
      - redis
      - chroma
    restart: always
  postgres:
    image: postgres:16
    volumes:
      - pgdata:/var/lib/postgresql/data
    environment:
      POSTGRES_DB: genai
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
  redis:
    image: redis:7-alpine
    volumes:
      - redisdata:/data
    command: redis-server --appendonly yes
  chroma:
    image: chromadb/chroma:latest
    volumes:
      - chromadata:/chroma/chroma
    ports:
      - "8001:8000"
volumes:
  pgdata:
  redisdata:
  chromadata:

Complete Import Reference for GenAI

# FastAPI Core
from fastapi import (
    FastAPI, APIRouter, Depends, HTTPException,
    Request, Response, BackgroundTasks, status,
    WebSocket, WebSocketDisconnect, Security,
)

# Parameters
from fastapi import Body, Form, File, UploadFile, Header, Cookie, Path, Query
# Responses - for streaming LLM output
from fastapi.responses import (
    JSONResponse, StreamingResponse, HTMLResponse,
    RedirectResponse, FileResponse, ORJSONResponse,
)
# Middleware - CORS for frontend, GZip for embeddings
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
# Security - API keys + OAuth2 + JWT
from fastapi.security import (
    OAuth2PasswordBearer, OAuth2PasswordRequestForm,
    APIKeyHeader, APIKeyQuery, HTTPBasic, HTTPBearer,
    SecurityScopes,
)
# Static files & templates - for chat UIs and report rendering
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
# Testing
from fastapi.testclient import TestClient
from fastapi.exceptions import RequestValidationError
# Pydantic - for all request/response models
from pydantic import BaseModel, Field, validator, root_validator
# GenAI stack
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
import chromadb
from sentence_transformers import SentenceTransformer
# Database
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
import redis.asyncio as redis
# Config
from pydantic_settings import BaseSettings

Building GenAI applications isn’t hard — but building production-ready GenAI applications takes intentional architecture.

FastAPI gives you the right building blocks: async routes for non-blocking LLM calls, StreamingResponse for UX-friendly token streaming, Depends() for sharing LLM clients without globals, Pydantic for validating the inputs your prompts depend on, and lifespan events for loading models and connections once at startup.

The biggest shift from a regular API to a GenAI API is thinking about failure modes: rate limits, context windows, content filters, slow model cold starts. FastAPI’s exception handlers give you the hooks to surface all of these cleanly to your clients.

Combine everything above with a vector database, a conversation memory layer, and a solid auth system, and you have a production-grade AI backend that can power anything from a simple chatbot to a multi-agent reasoning pipeline.

The prototype is the easy part. The API layer is where real products are made.

Built something cool with FastAPI + AI? The patterns above are the foundation. The rest is your imagination. 🚀