At JPMorgan Chase, before I moved into the training space, I spent two years building a high-concurrency API gateway that handled 40,000 requests per second. The lesson from that system that I apply directly to AI agent APIs: blocking I/O at scale is always fatal. Every synchronous call that waits for a slow downstream service becomes a thread that cannot serve other requests.

LLM calls are the slowest downstream services you will ever integrate. A gpt-4o call for a complex reasoning task can take 15–30 seconds. A RAG pipeline that retrieves 10 chunks and generates a 500-token response might take 8 seconds. If your AI agent API uses a synchronous framework like Flask, each of those calls occupies a worker thread for its entire duration. With 4 workers and 4 concurrent users all making 15-second LLM calls, your 5th user gets a timeout.

FastAPI solves this structurally. Its async model is not optional — it is the architecture. One async worker can handle hundreds of concurrent LLM calls by yielding CPU during each await, freeing the event loop to serve other requests while waiting for the LLM response. This is not a micro-optimisation; it is a 100x throughput difference at realistic AI agent concurrency levels.

Why FastAPI Is the Standard for AI Agent APIs

Feature Flask Django REST FastAPI
Async I/O ❌ Sync by default ⚠️ Partial (ASGI) ✅ First-class async
Streaming Responses ⚠️ Manual setup ⚠️ Complex ✅ StreamingResponse built-in
Request Validation ❌ Manual ⚠️ Serializers (verbose) ✅ Pydantic (auto)
OpenAPI Docs ❌ Plugin required ⚠️ DRF Spectacular ✅ Auto-generated
Concurrent LLM calls (4 workers) ~4 concurrent ~4–8 concurrent ~400+ concurrent

Async Endpoints: The Foundation

The single most impactful change you can make to an AI agent API: make every endpoint that calls an LLM async def and use await for every I/O operation. This is not optional for production systems.

Use the lifespan context manager (not the deprecated @app.on_event("startup")) for initialising shared resources like LLM clients, embedding models, and vector store connections. The lifespan pattern ensures resources are properly initialised before the first request and cleaned up when the process exits.

For timeout handling, wrap every LLM call in asyncio.wait_for with an explicit timeout. A hung LLM call that never returns will block an async worker indefinitely. The 30-second timeout is not a performance target — it is a safety net that prevents cascading failures when the upstream LLM API degrades.

Streaming Responses for Real-Time AI Output

Users abandon AI interfaces after 8 seconds of silence. Before streaming, a RAG query that takes 12 seconds shows nothing for 12 seconds, then the complete answer appears. With streaming, token 1 arrives in ~500ms and the user reads along as the model generates. Perceived quality increases dramatically even when total generation time is identical.

FastAPI's StreamingResponse with an async generator is the correct implementation. The generator yields chunks as they arrive from the LLM stream. The HTTP response stays open until the generator is exhausted. The media_type="text/event-stream" enables Server-Sent Events — a standard protocol that browsers and API clients handle natively with automatic reconnection.

For agent pipelines with multiple steps (retrieval → generation → validation), stream intermediate events too: "Retrieving context...", "Generating response...", "Validating output...". This transforms a 12-second black box into a transparent pipeline that users trust.

Health Check Patterns for Kubernetes

Kubernetes uses three probe types to manage pod lifecycle. Getting these wrong causes two failure modes: pods that are crashed but not restarted (liveness too permissive), or pods that restart endlessly due to transient dependency failures (liveness too strict). The correct pattern separates concerns clearly.

Probe Endpoint Checks Action on Failure
Liveness /health/live Process running, event loop active Restart pod
Readiness /health/ready LLM API reachable, vector store connected Remove from load balancer
Startup /health/startup Model loaded, warmup complete Delay liveness/readiness checks

Docker and Kubernetes Deployment

The Dockerfile for a FastAPI AI agent has two requirements beyond standard Python containers: a non-root user for security (Kubernetes PodSecurityPolicy will reject root containers in most enterprise environments), and a multi-stage build that separates dependency installation from the final image to minimise image size and attack surface.

Use uvicorn as the ASGI server with explicit --workers 2 for container deployments. Do not use Gunicorn + Uvicorn workers in containers — horizontal scaling via Kubernetes replicas is the correct scaling pattern, not vertical scaling via multiple workers per container. Each replica should be a lean, single-responsibility process.

Complete Production FastAPI Agent Code

# pip install fastapi uvicorn[standard] langchain-openai langchain-community pydantic

import asyncio
import logging
from contextlib import asynccontextmanager
from typing import AsyncGenerator

from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# -------------------------------------------------------------------
# Request / Response schemas
# WHY Pydantic: automatic validation + clear OpenAPI documentation.
# FastAPI rejects malformed requests before they reach your handler.
# -------------------------------------------------------------------
class AgentQuery(BaseModel):
    question: str = Field(..., min_length=1, max_length=4000, description="Question for the AI agent")
    session_id: str = Field(..., description="Session identifier for conversation context")
    stream: bool = Field(default=True, description="Stream tokens as they are generated")

class AgentResponse(BaseModel):
    answer: str
    session_id: str
    tokens_used: int
    latency_ms: float

class HealthStatus(BaseModel):
    status: str  # "healthy" | "degraded" | "unhealthy"
    checks: dict[str, str]

# -------------------------------------------------------------------
# Global state (initialised in lifespan, not at module level)
# WHY lifespan: replaces deprecated @app.on_event("startup").
# Resources are ready before the first request. Cleaned up on shutdown.
# -------------------------------------------------------------------
app_state: dict = {}

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator:
    # STARTUP — initialise all shared resources
    logger.info("Initialising LLM client...")
    app_state["llm"] = ChatOpenAI(
        model="gpt-4o-mini",
        temperature=0.0,
        streaming=True,
        request_timeout=30,  # seconds
        max_retries=2,
    )
    app_state["ready"] = True
    logger.info("Agent API ready")
    yield
    # SHUTDOWN — clean up resources
    app_state.clear()
    logger.info("Agent API shut down cleanly")

app = FastAPI(
    title="Gheware AI Agent API",
    version="1.0.0",
    description="Production-ready AI agent with streaming, health checks, and background tasks",
    lifespan=lifespan,
)

# -------------------------------------------------------------------
# Async non-streaming endpoint
# -------------------------------------------------------------------
@app.post("/agent/query", response_model=AgentResponse)
async def query_agent(request: AgentQuery, background_tasks: BackgroundTasks):
    """
    Non-streaming agent query. Returns the complete answer once generation finishes.
    Use for programmatic clients that need the full response before processing.
    """
    if not app_state.get("ready"):
        raise HTTPException(status_code=503, detail="Agent not yet initialised")

    llm: ChatOpenAI = app_state["llm"]
    start = asyncio.get_event_loop().time()

    try:
        # WHY asyncio.wait_for: prevents a hung LLM call from blocking the event loop forever.
        # 30s timeout matches the LLM client's request_timeout for consistent behaviour.
        response = await asyncio.wait_for(
            llm.ainvoke([HumanMessage(content=request.question)]),
            timeout=30.0,
        )
    except asyncio.TimeoutError:
        raise HTTPException(status_code=504, detail="LLM call timed out after 30 seconds")
    except Exception as e:
        logger.error(f"LLM error for session {request.session_id}: {e}")
        raise HTTPException(status_code=502, detail=f"LLM error: {str(e)}")

    latency_ms = (asyncio.get_event_loop().time() - start) * 1000
    tokens_used = response.usage_metadata.get("total_tokens", 0) if response.usage_metadata else 0

    # Background task: log interaction (does not affect response latency)
    background_tasks.add_task(log_interaction, request.session_id, request.question, latency_ms)

    return AgentResponse(
        answer=response.content,
        session_id=request.session_id,
        tokens_used=tokens_used,
        latency_ms=round(latency_ms, 1),
    )

# -------------------------------------------------------------------
# Streaming endpoint — Server-Sent Events
# -------------------------------------------------------------------
@app.post("/agent/stream")
async def stream_agent(request: AgentQuery):
    """
    Streaming agent query. Yields tokens via Server-Sent Events as they are generated.
    Use for chat UIs — eliminates perceived latency by showing token 1 within ~500ms.
    """
    if not app_state.get("ready"):
        raise HTTPException(status_code=503, detail="Agent not yet initialised")

    llm: ChatOpenAI = app_state["llm"]

    async def token_generator() -> AsyncGenerator[str, None]:
        try:
            async for chunk in llm.astream([HumanMessage(content=request.question)]):
                token = chunk.content
                if token:
                    # SSE format: "data: \n\n"
                    # WHY escaped newline: SSE spec requires double newline to delimit events.
                    yield f"data: {token}\n\n"
            # Signal stream completion to client
            yield "data: [DONE]\n\n"
        except Exception as e:
            logger.error(f"Streaming error: {e}")
            yield f"data: [ERROR] {str(e)}\n\n"

    return StreamingResponse(
        token_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",    # Disable Nginx buffering for true streaming
            "Connection": "keep-alive",
        },
    )

# -------------------------------------------------------------------
# Background task helper (fire and forget)
# -------------------------------------------------------------------
async def log_interaction(session_id: str, question: str, latency_ms: float) -> None:
    """Runs after the HTTP response is sent — does not block the client."""
    logger.info(f"Interaction logged | session={session_id} | latency={latency_ms:.0f}ms | q={question[:50]}")
# Health check endpoints + Dockerfile

# -------------------------------------------------------------------
# Three-tier health check endpoints
# -------------------------------------------------------------------
@app.get("/health/live", response_model=HealthStatus, tags=["Health"])
async def liveness():
    """
    Liveness probe — is the process running?
    NEVER include external dependency checks here.
    A failed LLM API call should NOT restart the pod — it should only
    remove it from the load balancer (readiness concern).
    """
    return HealthStatus(status="healthy", checks={"process": "alive"})

@app.get("/health/ready", response_model=HealthStatus, tags=["Health"])
async def readiness():
    """
    Readiness probe — can the pod serve traffic?
    Checks all dependencies. Failure removes pod from load balancer.
    """
    checks = {}
    overall = "healthy"

    # Check LLM connectivity (lightweight — just validate the client is configured)
    if app_state.get("ready") and app_state.get("llm"):
        checks["llm_client"] = "configured"
    else:
        checks["llm_client"] = "not_ready"
        overall = "unhealthy"

    if overall == "unhealthy":
        from fastapi.responses import JSONResponse
        return JSONResponse(
            status_code=503,
            content=HealthStatus(status=overall, checks=checks).model_dump()
        )
    return HealthStatus(status=overall, checks=checks)

@app.get("/health/startup", response_model=HealthStatus, tags=["Health"])
async def startup_probe():
    """
    Startup probe — has the application fully initialised?
    Kubernetes waits for this to succeed before enabling liveness/readiness.
    Gives slow-starting services (model loading, warm-up) extra time.
    """
    if app_state.get("ready"):
        return HealthStatus(status="healthy", checks={"init": "complete"})
    from fastapi.responses import JSONResponse
    return JSONResponse(
        status_code=503,
        content=HealthStatus(status="starting", checks={"init": "in_progress"}).model_dump()
    )

# -------------------------------------------------------------------
# Dockerfile (save as Dockerfile in project root)
# -------------------------------------------------------------------
# FROM python:3.11-slim AS builder
# WORKDIR /app
# COPY requirements.txt .
# RUN pip install --no-cache-dir --prefix=/install -r requirements.txt
#
# FROM python:3.11-slim
# WORKDIR /app
# COPY --from=builder /install /usr/local
# COPY . .
# # Run as non-root — required by most enterprise Kubernetes policies
# RUN adduser --disabled-password --gecos "" appuser && chown -R appuser /app
# USER appuser
# EXPOSE 8000
# # WHY --workers 2: single responsibility per container.
# # Scale horizontally via Kubernetes replicas, not vertical via workers.
# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

# -------------------------------------------------------------------
# Kubernetes Deployment snippet (k8s-deployment.yaml)
# -------------------------------------------------------------------
# apiVersion: apps/v1
# kind: Deployment
# spec:
#   replicas: 3
#   template:
#     spec:
#       containers:
#       - name: agent-api
#         image: ghcr.io/gheware/agent-api:latest
#         resources:
#           requests: {cpu: "250m", memory: "512Mi"}
#           limits:   {cpu: "1000m", memory: "1Gi"}
#         startupProbe:
#           httpGet: {path: /health/startup, port: 8000}
#           failureThreshold: 30     # 30 * 10s = 5 min startup budget
#           periodSeconds: 10
#         livenessProbe:
#           httpGet: {path: /health/live, port: 8000}
#           initialDelaySeconds: 5
#           periodSeconds: 15
#           failureThreshold: 3
#         readinessProbe:
#           httpGet: {path: /health/ready, port: 8000}
#           initialDelaySeconds: 10
#           periodSeconds: 10
#           failureThreshold: 3

Frequently Asked Questions

Why use FastAPI for AI agents instead of Flask or Django?

FastAPI is purpose-built for async I/O. LLM calls take 2–30 seconds each. Flask's synchronous model blocks a worker thread for the entire duration of each LLM call — under 10 concurrent users, every worker is occupied. FastAPI's async model handles hundreds of concurrent LLM calls with a single worker by yielding control during each await. Additionally: auto-generated OpenAPI docs, built-in Pydantic validation, and native streaming support via StreamingResponse.

How do I stream LLM responses with FastAPI?

Use StreamingResponse with an async generator that yields Server-Sent Event formatted strings (data: {chunk}\n\n). The LLM's astream() method returns an async iterator of chunks. Yield each non-empty chunk immediately. Signal completion with data: [DONE]\n\n. Set media_type="text/event-stream" and the X-Accel-Buffering: no header to disable Nginx buffering in production deployments.

How many Uvicorn workers should I run for an AI agent API?

For containerised deployments: 1–2 workers per container. LLM calls are network-bound, not CPU-bound — a single async worker handles hundreds of concurrent LLM calls by yielding during each await. More workers per container add memory overhead without proportional throughput gain. Scale horizontally via Kubernetes replicas instead. Start with 2 workers, measure p95 latency under load at your expected concurrency, add workers only if CPU utilisation consistently exceeds 70%.

What health check endpoints should my AI agent API expose?

Three endpoints: /health/live (liveness — is the process alive? no external checks, returns 200 if the event loop is running), /health/ready (readiness — are all dependencies reachable? checks LLM API connectivity and vector store, returns 503 to remove from load balancer if any check fails), and /health/startup (startup — has model loading completed? gives slow-starting services extra time before liveness checks begin). Never put dependency checks in liveness — a temporary LLM API outage should not restart your pod.

Conclusion: FastAPI Is Infrastructure, Not Scaffolding

The teams I see struggle with AI agent API performance are consistently the ones that treat FastAPI as a thin HTTP wrapper — a place to put their agent logic. The teams that succeed treat FastAPI as infrastructure: they design for async from the first line, they stream every response, they implement health checks for every deployment environment, and they use background tasks for anything that does not affect the HTTP response.

Day 3 of the Agentic AI Workshop covers the full FastAPI deployment pipeline — from the first async def endpoint through to a running Kubernetes deployment with three-tier health probes and Prometheus metrics scraping. Every participant deploys their own agent API on a real cluster by end of day. The difference between understanding these patterns and having shipped them is exactly one lab session.