FastAPI Agent APIs

The Agent API Architecture

flowchart LR
    subgraph EP["Endpoints"]
        R1["POST /agent/run\nsync, short tasks"]
        R2["POST /agent/start\nasync, returns task_id"]
        R3["GET /agent/status\npoll task status"]
        R4["GET /agent/stream\nSSE token streaming"]
    end
    subgraph Infra["Infrastructure"]
        A["Auth + Rate\nLimiting"]
        Q[("Task Queue\nRedis")]
        W["Background\nWorkers"]
    end
    EP --> A --> Q --> W
    style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Q fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style W fill:#fff7ed,stroke:#f59e0b,color:#0F172A

Complete FastAPI Agent Service

# src/api/main.py
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, Header
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from typing import Optional
import uuid, json, asyncio
from datetime import datetime
from contextlib import asynccontextmanager

# Request/Response schemas: FastAPI reads these to validate requests and generate OpenAPI docs
class AgentRunRequest(BaseModel):
    query: str = Field(min_length=1, max_length=10000)  # reject empty or oversized queries automatically
    thread_id: Optional[str] = None  # if provided, the agent continues an existing conversation
    model: str = Field(default="gpt-5.6-sol", pattern="^(gpt-5.6-sol|gpt-5.6-luna|claude-sonnet-5)$")  # allowlist via regex
    max_iterations: int = Field(default=20, ge=1, le=50)  # ge/le enforce min/max without custom validators
    stream: bool = Field(default=False)

class AgentRunResponse(BaseModel):
    run_id: str
    status: str  # pending | running | completed | failed
    result: Optional[str] = None
    error: Optional[str] = None
    usage: Optional[dict] = None
    created_at: str
    completed_at: Optional[str] = None

class TaskStatusResponse(BaseModel):
    run_id: str
    status: str
    progress: Optional[str] = None
    result: Optional[str] = None
    error: Optional[str] = None

# In-memory task store (use Redis in production)
# keyed by run_id so any worker can look up any task status
task_store: dict[str, dict] = {}

# lifespan context manager runs setup/teardown once at server start/stop: not per request
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    await memory_store.initialize()
    yield
    # Shutdown: clean up resources

app = FastAPI(
    title="AI Agent API",
    version="1.0.0",
    lifespan=lifespan,  # pass lifespan so FastAPI runs startup/shutdown hooks
)

# CORS middleware allows browser clients from your frontend domain to call this API
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://your-frontend.com"],  # restrict to your own domain in production
    allow_methods=["*"],
    allow_headers=["*"],
)

Authentication & Rate Limiting

# src/api/auth.py
from fastapi import HTTPException, Depends, Header
from functools import lru_cache
import hashlib

# API key store (use a database in production)
# storing hashes instead of raw keys: if this dict leaks, keys can't be used directly
API_KEYS = {
    hashlib.sha256("sk-prod-xxxx".encode()).hexdigest(): {
        "tier": "premium",
        "rate_limit_per_min": 60,    # premium tier gets 60 requests per minute
        "cost_limit_usd_per_day": 50.0,
    },
    hashlib.sha256("sk-free-yyyy".encode()).hexdigest(): {
        "tier": "free",
        "rate_limit_per_min": 10,    # free tier is throttled more aggressively
        "cost_limit_usd_per_day": 1.0,
    },
}

# Header(...) makes X-API-Key a required header: FastAPI returns 422 if it's missing
def get_api_key(x_api_key: str = Header(..., alias="X-API-Key")) -> dict:
    """Validate API key and return key metadata."""
    # hash the incoming key before lookup so raw keys never touch our comparisons
    key_hash = hashlib.sha256(x_api_key.encode()).hexdigest()
    
    if key_hash not in API_KEYS:
        raise HTTPException(status_code=401, detail="Invalid API key")
    
    return API_KEYS[key_hash]

# Rate limiting with in-memory sliding window
from collections import defaultdict, deque
import time

# defaultdict(deque) auto-creates an empty deque for each new key_id on first access
request_windows: dict[str, deque] = defaultdict(deque)

# Depends(get_api_key) runs get_api_key first and injects the result: dependency injection
def check_rate_limit(api_key_meta: dict = Depends(get_api_key)) -> dict:
    key_id = api_key_meta.get("tier", "unknown")
    limit = api_key_meta["rate_limit_per_min"]
    window = request_windows[key_id]
    
    now = time.time()
    # Remove requests older than 60 seconds: sliding window only counts the last minute
    while window and window[0] < now - 60:
        window.popleft()
    
    if len(window) >= limit:
        raise HTTPException(
            status_code=429,  # 429 Too Many Requests is the standard rate-limit status code
            detail=f"Rate limit exceeded ({limit} req/min). Try again shortly.",
            headers={"Retry-After": "10"},  # tell the client to wait 10 seconds before retrying
        )
    
    window.append(now)  # record this request's timestamp in the sliding window
    return api_key_meta

Async Background Task Endpoints

# src/api/endpoints.py

# response_model=AgentRunResponse tells FastAPI to validate and serialize the return value
@app.post("/agent/start", response_model=AgentRunResponse)
async def start_agent_run(
    request: AgentRunRequest,
    background_tasks: BackgroundTasks,  # FastAPI injects this; used to schedule async work
    api_meta: dict = Depends(check_rate_limit),  # runs auth + rate limit before the handler
):
    """Start an agent run asynchronously. Returns immediately with a run_id to poll."""
    run_id = str(uuid.uuid4())  # universally unique ID so clients can poll the right task
    thread_id = request.thread_id or run_id  # use provided thread_id for conversation continuity
    
    # record "pending" before adding the task so polling can start immediately
    task_store[run_id] = {
        "status": "pending",
        "progress": "Initializing agent...",
        "created_at": datetime.utcnow().isoformat(),
    }
    
    async def run_agent_task():
        task_store[run_id]["status"] = "running"
        try:
            config = {"configurable": {"thread_id": thread_id}}  # thread_id routes to the right checkpointer memory
            result = await agent.ainvoke(
                {"messages": [HumanMessage(content=request.query)], "iterations": 0, "final_answer": None},
                config=config,
            )
            # filter to only AI messages: ignore ToolMessages and HumanMessages in the output
            ai_msgs = [m for m in result["messages"] if isinstance(m, AIMessage)]
            
            task_store[run_id].update({
                "status": "completed",
                "result": ai_msgs[-1].content if ai_msgs else "",
                "completed_at": datetime.utcnow().isoformat(),
            })
        except Exception as e:
            # catch all exceptions so a crashing agent doesn't leave the task stuck in "running"
            task_store[run_id].update({
                "status": "failed",
                "error": str(e),
                "completed_at": datetime.utcnow().isoformat(),
            })
    
    # background_tasks.add_task() returns immediately: the agent runs in a separate coroutine
    background_tasks.add_task(run_agent_task)
    
    # return "pending" right away so the client gets a run_id to poll
    return AgentRunResponse(
        run_id=run_id,
        status="pending",
        created_at=task_store[run_id]["created_at"],
    )

@app.get("/agent/status/{run_id}", response_model=TaskStatusResponse)
async def get_run_status(run_id: str, api_meta: dict = Depends(check_rate_limit)):
    """Poll the status of a background agent run."""
    if run_id not in task_store:
        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
    
    task = task_store[run_id]
    return TaskStatusResponse(
        run_id=run_id,
        status=task["status"],
        progress=task.get("progress"),  # optional, may be None if progress tracking isn't set
        result=task.get("result"),
        error=task.get("error"),
    )

@app.post("/agent/run", response_model=AgentRunResponse)
async def run_agent_sync(
    request: AgentRunRequest,
    api_meta: dict = Depends(check_rate_limit),
):
    """Synchronous agent run (for quick tasks only, max 30s)."""
    if request.max_iterations > 10:
        # enforce a lower iteration cap on the sync endpoint to stay under load-balancer timeouts
        raise HTTPException(
            status_code=400,
            detail="Sync endpoint limited to max_iterations=10. Use /agent/start for longer runs."
        )
    
    try:
        # asyncio.wait_for() enforces a hard timeout: raises TimeoutError if agent exceeds 30 seconds
        result = await asyncio.wait_for(
            agent.ainvoke(
                {"messages": [HumanMessage(content=request.query)], "iterations": 0, "final_answer": None}
            ),
            timeout=30.0,
        )
        
        ai_msgs = [m for m in result["messages"] if isinstance(m, AIMessage)]
        return AgentRunResponse(
            run_id=str(uuid.uuid4()),
            status="completed",
            result=ai_msgs[-1].content if ai_msgs else "",
            created_at=datetime.utcnow().isoformat(),
            completed_at=datetime.utcnow().isoformat(),
        )
    
    except asyncio.TimeoutError:
        # return 504 Gateway Timeout: the standard HTTP code for a timed-out upstream service
        raise HTTPException(status_code=504, detail="Agent run timed out (30s). Use /agent/start.")

Health Check & Metrics

@app.get("/health")
async def health_check():
    """Liveness check for load balancer."""
    # liveness endpoint: if this returns 200, the container is alive (but not necessarily ready)
    return {"status": "ok", "timestamp": datetime.utcnow().isoformat()}

@app.get("/health/ready")
async def readiness_check():
    """Readiness check, returns 503 if dependencies are down."""
    # readiness endpoint: Kubernetes removes the pod from the load balancer if this fails
    checks = {}
    
    try:
        # actively probe the OpenAI API: a 401 or timeout here means the pod can't serve requests
        await openai_client.models.list()
        checks["openai"] = "ok"
    except Exception as e:
        checks["openai"] = f"error: {e}"
    
    all_ok = all(v == "ok" for v in checks.values())
    # return 503 if any dependency is unhealthy so orchestrators stop routing traffic here
    return {"status": "ready" if all_ok else "degraded", "checks": checks}

@app.get("/metrics")
async def get_metrics(api_meta: dict = Depends(check_rate_limit)):
    """Basic operational metrics."""
    # count tasks by status: useful for dashboards and alerting on failed run rates
    statuses = [t["status"] for t in task_store.values()]
    return {
        "total_runs": len(task_store),
        "pending": statuses.count("pending"),
        "running": statuses.count("running"),
        "completed": statuses.count("completed"),
        "failed": statuses.count("failed"),  # a rising failed count is the first sign of trouble
    }

Running the Service

# Development: --reload watches for file changes and restarts automatically
uvicorn src.api.main:app --reload --port 8000

# Production: --workers 4 spawns 4 processes; uvloop is a faster event loop for Linux
uvicorn src.api.main:app --host 0.0.0.0 --port 8000 --workers 4 --loop uvloop

With multiple workers, ensure your agent state store (LangGraph checkpointer) uses a shared backend (Postgres, Redis) rather than in-memory, so all worker processes access the same state.