2. Deployment patterns for agents

Description

In the previous capsule you saw the gap between prototype and production: concurrency, latency, cost, observability, resilience. All of that sounds abstract until you need to put your agent in front of real users. This capsule is where it stops being abstract — we're going to deploy a LangGraph agent with FastAPI, containerize it with Docker, set up health checks that actually catch problems, and handle shutdown without dropping in-flight requests.

The most common mistake I see in teams shipping agents is treating deployment as uvicorn main:app and calling it a day. An agent is not a CRUD app. An agent makes LLM calls that take 5-30 seconds, invokes external tools that can fail, keeps state between turns, and consumes resources unpredictably. Trivial deployment is a mistake you pay for with downtime, lost data, and surprise bills.

Connection to the module: This capsule sets the base infrastructure. First we deploy properly (capsule 02), then we scale (03), monitor (04), handle errors (05), and control costs (06). Without a solid deployment, scaling and monitoring have no foundation.

FastAPI + LangGraph: the deployment stack

Why FastAPI? Three reasons: native async (critical for agents that wait on LLM responses), automatic documentation with OpenAPI (your endpoints document themselves), and the most mature Python ecosystem for HTTP APIs.

The minimum viable server

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from contextlib import asynccontextmanager
import uuid
import logging

from agent import build_research_agent

logger = logging.getLogger(__name__)

checkpointer = MemorySaver()
graph = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global graph
    logger.info("Compiling agent...")
    agent = build_research_agent()
    graph = agent.compile(checkpointer=checkpointer)
    logger.info("Agent ready")
    yield
    logger.info("Shutdown: cleaning up resources...")

app = FastAPI(
    title="Research Agent API",
    description="REST API for the production Research Agent",
    version="1.0.0",
    lifespan=lifespan,
)

Stop at lifespan. FastAPI has two ways to handle startup/shutdown: the @app.on_event decorators (deprecated) and the lifespan async context manager (the right way). The yield splits startup from shutdown — everything before the yield runs on boot, everything after runs on shutdown. We compile the graph once at startup, not on every request.

Request/response models

class ChatRequest(BaseModel):
    message: str
    thread_id: str | None = None

class ChatResponse(BaseModel):
    response: str
    thread_id: str
    metadata: dict | None = None

class AsyncJobResponse(BaseModel):
    job_id: str
    status: str
    thread_id: str

The synchronous endpoint

The synchronous endpoint is the simplest — the client waits until the agent answers:

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    thread_id = request.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    try:
        result = await graph.ainvoke(
            {"messages": [HumanMessage(content=request.message)]},
            config,
        )
        return ChatResponse(
            response=result["messages"][-1].content,
            thread_id=thread_id,
            metadata={"steps": len(result["messages"])},
        )
    except Exception as e:
        logger.error(f"Error in chat: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail="Error processing the request")

Notice the ainvoke — the async version of invoke. If you use invoke (synchronous) inside an async def endpoint, you block FastAPI's event loop and a single slow request freezes the whole server. ainvoke releases the event loop while it waits for the LLM's response.

The asynchronous endpoint: for long tasks

A research query can take 30-60 seconds. No client should wait that long with the connection open. The pattern: accept the job, return an ID immediately, process in the background:

jobs: dict[str, dict] = {}

async def run_agent_job(job_id: str, message: str, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    jobs[job_id]["status"] = "running"

    try:
        result = await graph.ainvoke(
            {"messages": [HumanMessage(content=message)]},
            config,
        )
        jobs[job_id]["status"] = "completed"
        jobs[job_id]["result"] = result["messages"][-1].content
    except Exception as e:
        jobs[job_id]["status"] = "failed"
        jobs[job_id]["error"] = str(e)
        logger.error(f"Job {job_id} failed: {e}", exc_info=True)

@app.post("/chat/async", response_model=AsyncJobResponse)
async def chat_async(request: ChatRequest, background_tasks: BackgroundTasks):
    job_id = str(uuid.uuid4())
    thread_id = request.thread_id or str(uuid.uuid4())

    jobs[job_id] = {"status": "pending", "thread_id": thread_id}
    background_tasks.add_task(run_agent_job, job_id, request.message, thread_id)

    return AsyncJobResponse(
        job_id=job_id, status="pending", thread_id=thread_id,
    )

@app.get("/jobs/{job_id}")
async def get_job_status(job_id: str):
    if job_id not in jobs:
        raise HTTPException(status_code=404, detail="Job not found")
    return jobs[job_id]

This pattern (submit → poll → get result) is the industry standard for long operations. In real production, replace the jobs dictionary with Redis or a database — the in-memory dict is lost if the server restarts.

Automatic documentation with OpenAPI

FastAPI generates OpenAPI documentation automatically from the Pydantic models. Visit http://localhost:8000/docs (Swagger UI) and you get an interactive interface where any developer on your team can try the agent without writing code. Add metadata to improve the documentation:

@app.post(
    "/chat",
    response_model=ChatResponse,
    summary="Synchronous chat with the agent",
    description="Send a message and wait for the response. For long queries, use /chat/async.",
    tags=["Agent"],
)
async def chat(request: ChatRequest):
    ...

Containerization with Docker

Your agent works on your machine. Now it needs to work on any machine. Docker packages your application with all its dependencies into a reproducible image.

Dockerfile: multi-stage build

# === Stage 1: Builder ===
FROM python:3.12-slim AS builder

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# === Stage 2: Runtime ===
FROM python:3.12-slim AS runtime

WORKDIR /app

RUN groupadd -r agent && useradd -r -g agent agent

COPY --from=builder /install /usr/local
COPY src/ ./src/
COPY main.py .
RUN chown -R agent:agent /app
USER agent

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]

The key decisions:

• Multi-stage: The builder stage installs the build dependencies. The runtime stage only copies the packages. Result: an image 70% smaller.
• Non-root user: The agent user is basic security practice. If someone compromises the container, they don't get root.
• --workers 1: LangGraph agents use a lot of memory (checkpoints, tool state). Multiple workers multiply the consumption. Scale with multiple containers, not multiple workers.
• HEALTHCHECK: Docker runs it every 30 seconds. If it fails 3 times, the container is marked unhealthy. Orchestrators like Kubernetes restart it.

Docker Compose

Your agent doesn't live alone. It needs Redis for caching and possibly PostgreSQL for checkpoints:

services:
  agent:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - LANGSMITH_API_KEY=${LANGSMITH_API_KEY}
      - LANGSMITH_TRACING=true
      - REDIS_URL=redis://redis:6379
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped
    deploy:
      resources:
        limits:
          memory: 2G

  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
    volumes:
      - redis_data:/data

volumes:
  redis_data:

• depends_on with condition: service_healthy: The agent doesn't boot until Redis is healthy.
• restart: unless-stopped: If the container crashes, Docker restarts it automatically.
• memory: 2G: Prevents a runaway agent from eating all the host's resources.
• Secrets via environment: API keys come from environment variables, never hardcoded.

Don't forget the .dockerignore — without it, Docker copies .git, __pycache__, .env, and test files into the build context:

.git
.env
.env.*
__pycache__
*.pyc
.pytest_cache
tests/
*.md

Health checks: verify that everything works

A /health endpoint that always returns {"status": "ok"} is useless. Your agent depends on an external LLM, on MCP servers, on valid API keys, and possibly on a database. A real health check verifies every dependency.

A production health check

from datetime import datetime, timezone
import os
import httpx

@app.get("/health", tags=["System"])
async def health_check():
    checks = {}
    overall_healthy = True

    checks["agent"] = {
        "status": "healthy" if graph is not None else "unhealthy",
    }

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            response = await client.get(
                "https://api.openai.com/v1/models",
                headers={"Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}"},
            )
            checks["openai"] = {
                "status": "healthy" if response.status_code == 200 else "degraded",
            }
    except Exception as e:
        checks["openai"] = {"status": "unhealthy", "detail": str(e)}
        overall_healthy = False

    try:
        import redis.asyncio as aioredis
        r = aioredis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))
        await r.ping()
        checks["redis"] = {"status": "healthy"}
        await r.aclose()
    except Exception as e:
        checks["redis"] = {"status": "unhealthy", "detail": str(e)}
        overall_healthy = False

    required_keys = ["OPENAI_API_KEY", "LANGSMITH_API_KEY"]
    missing = [k for k in required_keys if not os.getenv(k)]
    checks["api_keys"] = {
        "status": "healthy" if not missing else "unhealthy",
        "detail": f"Missing: {missing}" if missing else "All configured",
    }
    if missing:
        overall_healthy = False

    return {
        "status": "healthy" if overall_healthy else "unhealthy",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "version": os.getenv("APP_VERSION", "dev"),
        "checks": checks,
    }

Three levels of health check

In production with Kubernetes, you need three levels:

@app.get("/health/live")
async def liveness():
    return {"status": "alive"}

@app.get("/health/ready")
async def readiness():
    if graph is None:
        raise HTTPException(status_code=503, detail="Agent not ready")
    return {"status": "ready"}

Liveness is trivial — if the process answers HTTP, it's alive. Readiness verifies the agent can process requests. Startup is like readiness but is only checked during boot — an agent can take 30+ seconds to compile the graph.

LangGraph Platform: managed deployment

LangGraph Platform is LangChain's managed service for deploying LangGraph agents. Instead of you handling FastAPI, Docker, scaling, and monitoring, LangGraph Platform takes care of the infrastructure and you just ship the graph.

How it works

1. You define your agent in a standard Python file with LangGraph
2. You create a langgraph.json with the deployment configuration
3. You deploy with langgraph deploy or from the LangSmith dashboard
4. You consume your agent via the REST API that LangGraph Platform generates

{
    "dependencies": ["./requirements.txt"],
    "graphs": {
        "research_agent": "./agent.py:graph"
    },
    "env": ".env"
}

What it includes

• Hosting and automatic scaling — you don't manage servers
• Persistent checkpointing — conversation state saved automatically
• Native streaming — support for token and event streaming
• LangSmith built in — tracing, evaluation, and monitoring with no setup
• Double-texting handling — native handling of concurrent user messages

When to pick LangGraph Platform vs self-hosted

LangGraph Platform is better when:

• Small team with no dedicated DevOps
• Aggressive time-to-market — you need to deploy in hours
• Deep integration with LangSmith

Self-hosted is better when:

• Total control is required (regulation, compliance, data residency)
• Predictable, high scale (you optimize cost per request)
• Vendor lock-in is unacceptable

Environment configuration

Your agent needs configuration that varies across environments. API keys, service URLs, feature flags, log levels. The golden rule: never hardcode configuration in the code.

Configuration with Pydantic Settings

from pydantic_settings import BaseSettings
from pydantic import Field

class Settings(BaseSettings):
    openai_api_key: str = Field(..., description="OpenAI API key")
    langsmith_api_key: str = Field("", description="LangSmith API key")

    model_name: str = Field("gpt-4.1-mini", description="LLM model to use")
    max_tokens: int = Field(4096, description="Max tokens per response")
    temperature: float = Field(0.0, description="LLM temperature")

    redis_url: str = Field("redis://localhost:6379", description="Redis URL")
    log_level: str = Field("INFO", description="Logging level")

    enable_tracing: bool = Field(True, description="Enable LangSmith tracing")
    max_agent_steps: int = Field(25, description="Max steps before forced stop")
    request_timeout: int = Field(120, description="Request timeout in seconds")

    model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}

settings = Settings()

pydantic_settings gives you: validation at startup (missing openai_api_key → the app doesn't boot), type coercion (string from the environment → int), and sensible defaults. Failing fast beats failing on the first request.

Secrets management

The rule: secrets never touch the repository. Add .env to .gitignore on day one.

Configuration per environment

# .env.development
MODEL_NAME=gpt-4.1-mini
LOG_LEVEL=DEBUG
ENABLE_TRACING=false

# .env.production
MODEL_NAME=gpt-4.1
LOG_LEVEL=WARNING
ENABLE_TRACING=true

In development you use the cheap model with verbose logging. In production, the full model with tracing enabled. Same application, different behavior — controlled by configuration, not by code.

Graceful shutdown: don't drop in-flight requests

When you deploy a new version, the old server has to shut down. If you kill it with kill -9, any in-flight request is lost — including agents halfway through a 45-second investigation. Graceful shutdown: stop accepting new requests, wait for the in-flight ones to finish, clean up resources, and shut down.

import asyncio

active_requests: set[str] = set()
shutdown_event = asyncio.Event()

@asynccontextmanager
async def lifespan(app: FastAPI):
    global graph
    agent = build_research_agent()
    graph = agent.compile(checkpointer=checkpointer)
    logger.info("Agent compiled and ready")

    yield

    logger.info("Starting graceful shutdown...")
    shutdown_event.set()

    if active_requests:
        logger.info(f"Waiting on {len(active_requests)} in-flight requests...")
        timeout = 60
        start = asyncio.get_event_loop().time()
        while active_requests:
            elapsed = asyncio.get_event_loop().time() - start
            if elapsed > timeout:
                logger.warning(
                    f"Shutdown timeout. Cancelling {len(active_requests)} requests."
                )
                break
            await asyncio.sleep(0.5)

    logger.info("Shutdown complete")

@app.middleware("http")
async def track_requests(request, call_next):
    if shutdown_event.is_set():
        raise HTTPException(status_code=503, detail="Server shutting down")

    request_id = str(uuid.uuid4())
    active_requests.add(request_id)
    try:
        response = await call_next(request)
        return response
    finally:
        active_requests.discard(request_id)

The flow: the shutdown signal arrives → shutdown_event fires → the middleware rejects new requests with a 503 → it waits on in-flight requests for up to 60 seconds → a safety timeout cancels whatever is left → cleanup and shutdown.

With LangGraph's persistent checkpointing (PostgreSQL, Redis), the agent's state is already saved after every step. Another instance can pick the conversation back up exactly where it stopped.

Comparison: self-hosted vs LangGraph Platform

A pragmatic recommendation: Start with LangGraph Platform to validate your agent with real users. If traction confirms the product has a future, migrate to self-hosted when you need control, scale, or optimized cost. Don't spend weeks on infrastructure for an agent nobody is going to use.

Connection to the project

The Research Agent you've been building across M4-M9 is about to become a production service using these patterns:

1. A FastAPI server with three endpoints: /chat for quick queries, /chat/async for long investigations, and /jobs/{id} to check status.
2. A Docker container with a multi-stage build, a non-root user, and a configured healthcheck.
3. Health checks that verify: the agent is compiled, OpenAI is reachable, Redis is connected, API keys are present.
4. Environment config with pydantic_settings — API keys, model names, and feature flags come from the environment.
5. Graceful shutdown that waits on in-flight requests and saves the state of incomplete jobs.

In capsule 03 (Scaling), you'll add caching with Redis and async patterns. In 04 (Monitoring), you'll connect LangSmith and configure alerts. Each capsule adds a layer on top of this base.

Troubleshooting

Problem 1: invoke blocks the event loop

Symptom: The server answers one request at a time. The second request waits until the first one finishes.

Cause: You're using graph.invoke() (synchronous) inside an async def endpoint. It blocks the event loop — FastAPI can't process other requests.

Fix: Use await graph.ainvoke(). If you need invoke, declare the endpoint as def (without async) so FastAPI runs it in a thread pool.

Problem 2: the container runs out of memory

Symptom: The container gets killed by the OOM killer after a few hours. Logs show Killed with no detail.

Cause: MemorySaver stores checkpoints in memory. Every conversation adds state. After thousands of conversations, memory grows past the limit.

Fix: Use a persistent checkpointer in production:

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

checkpointer = AsyncPostgresSaver.from_conn_string(os.getenv("DATABASE_URL"))

Problem 3: the health check passes but the agent doesn't answer

Symptom: /health returns 200 but /chat fails with a timeout. The load balancer keeps sending traffic.

Cause: The health check only verifies the process answers HTTP, not that the agent can process requests. The API key expired, or the LLM rate limit was hit.

Fix: Implement the complete health check that verifies every dependency. If OpenAI fails, return 503 so the load balancer stops sending traffic.

Problem 4: requests lost during the deploy

Symptom: Intermittent errors during deploys. In-flight requests get cut off when the old container shuts down.

Cause: Docker sends SIGTERM and waits 10 seconds by default. If your agent takes longer, Docker sends SIGKILL.

Fix: Raise stop_grace_period and implement graceful shutdown:

services:
  agent:
    stop_grace_period: 120s

Problem 5: environment variables don't reach the container

Symptom: A Pydantic validation error at boot: openai_api_key field required. The variables are in .env but the container doesn't see them.

Cause: Docker Compose needs an explicit env_file. And careful: Docker reads quotes literally.

# Correct
OPENAI_API_KEY=sk-abc123

# Wrong (Docker includes the quotes as part of the value)
OPENAI_API_KEY="sk-abc123"

Exercises

Exercise 1: Design the endpoints

Your team needs to expose a technical support agent as an API. It takes customer questions, searches the knowledge base, and answers. Conversations have to persist. Design the REST endpoints: paths, HTTP methods, request/response bodies, and status codes.

POST /conversations
  Request:  { "user_id": "user-123", "initial_message": "My app can't connect to the DB" }
  Response: { "conversation_id": "conv-abc", "response": "...", "sources": [...] }
  Status:   201 Created

POST /conversations/{conversation_id}/messages
  Request:  { "message": "I already checked the connection string" }
  Response: { "response": "...", "sources": [...] }
  Status:   200 OK

GET /conversations/{conversation_id}
  Response: { "conversation_id": "...", "messages": [...], "created_at": "..." }
  Status:   200 OK

GET /health
  Response: { "status": "healthy", "checks": {...} }
  Status:   200 OK / 503 Service Unavailable

Exercise 2: Fix the Dockerfile

This Dockerfile has 5 production problems. Find them:

FROM python:3.12
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt
ENV OPENAI_API_KEY=sk-abc123mykey
CMD python main.py

FROM python:3.12-slim
WORKDIR /app
RUN groupadd -r agent && useradd -r -g agent agent
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN chown -R agent:agent /app
USER agent
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Exercise 3: A complete health check

Write a health check for an agent that depends on: the OpenAI API, PostgreSQL, Redis, and LangSmith. It should return the individual status of each dependency plus an overall status. LangSmith is non-critical (degraded if it fails, not unhealthy).

@app.get("/health")
async def health():
    checks = {}
    healthy = True

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            r = await client.get(
                "https://api.openai.com/v1/models",
                headers={"Authorization": f"Bearer {settings.openai_api_key}"},
            )
            checks["openai"] = {"status": "healthy" if r.status_code == 200 else "degraded"}
    except Exception as e:
        checks["openai"] = {"status": "unhealthy", "error": str(e)}
        healthy = False

    try:
        import asyncpg
        conn = await asyncpg.connect(settings.database_url, timeout=5)
        await conn.execute("SELECT 1")
        await conn.close()
        checks["postgres"] = {"status": "healthy"}
    except Exception as e:
        checks["postgres"] = {"status": "unhealthy", "error": str(e)}
        healthy = False

    try:
        import redis.asyncio as aioredis
        r = aioredis.from_url(settings.redis_url)
        await r.ping()
        await r.aclose()
        checks["redis"] = {"status": "healthy"}
    except Exception as e:
        checks["redis"] = {"status": "unhealthy", "error": str(e)}
        healthy = False

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            r = await client.get(
                "https://api.smith.langchain.com/info",
                headers={"x-api-key": settings.langsmith_api_key},
            )
            checks["langsmith"] = {"status": "healthy" if r.status_code == 200 else "degraded"}
    except Exception as e:
        checks["langsmith"] = {"status": "degraded", "error": str(e)}

    return {"status": "healthy" if healthy else "unhealthy", "checks": checks}

Exercise 4: Graceful shutdown

Your agent processes queries that take 30-90 seconds. Implement a graceful shutdown that: (1) rejects new requests, (2) waits up to 120 seconds for in-flight ones, (3) logs shutdown metrics.

import time

shutdown_requested = False
active_tasks: dict[str, dict] = {}

@app.middleware("http")
async def shutdown_middleware(request, call_next):
    if shutdown_requested and request.url.path not in ("/health", "/health/live"):
        return JSONResponse(status_code=503, content={"detail": "Shutting down"})

    task_id = str(uuid.uuid4())
    active_tasks[task_id] = {"path": request.url.path, "started_at": time.time()}
    try:
        response = await call_next(request)
        return response
    finally:
        active_tasks.pop(task_id, None)

@asynccontextmanager
async def lifespan(app: FastAPI):
    global graph, shutdown_requested
    graph = build_research_agent().compile(checkpointer=checkpointer)
    yield

    shutdown_requested = True
    start = time.time()
    while active_tasks and (time.time() - start) < 120:
        logger.info(f"Shutdown: waiting on {len(active_tasks)} requests...")
        await asyncio.sleep(2)

    if active_tasks:
        logger.warning(f"Forced shutdown. {len(active_tasks)} requests not completed.")
    logger.info(f"Shutdown complete in {time.time() - start:.1f}s")

Exercise 5: Self-hosted vs Platform

Your startup has a customer support agent: ~500 queries/day, a team of 3 developers with no DevOps. Self-hosted or LangGraph Platform? In 6 months you grow to 50,000 queries/day and hire DevOps. Does your recommendation change?

Summary

• FastAPI + LangGraph is the deployment stack. REST endpoints with ainvoke so nothing blocks. Synchronous for quick queries, asynchronous (submit → poll) for long investigations.
• Docker with multi-stage builds shrinks the image. Non-root user, embedded healthcheck, Docker Compose to orchestrate dependencies.
• Production health checks verify every dependency: agent, OpenAI, Redis, MCP servers, API keys. Three levels: liveness, readiness, startup.
• LangGraph Platform is the managed option: deploy in minutes, automatic scaling, LangSmith included. Self-hosted when you need control, scale, or compliance.
• Environment configuration with pydantic_settings validates at startup. Secrets via environment variables, never hardcoded.
• Graceful shutdown rejects new requests, waits on in-flight ones with a timeout, and saves state. Without it, every deploy drops requests.
• Self-hosted vs managed depends on context: team size, volume, budget, compliance. There's no universal answer.

Next capsule: Scaling and Performance — async execution, connection pooling, caching strategies, and horizontal scaling patterns.

Additional resources

1. FastAPI Documentation — Official documentation with tutorials and API reference
2. LangGraph Platform — Quick start for LangGraph's managed service
3. Docker Multi-Stage Builds — Official documentation on multi-stage builds
4. 12-Factor App — Methodology for cloud-native applications
5. Kubernetes Health Probes — Liveness, readiness, and startup probes
6. Pydantic Settings — Configuration management with type validation