How to Build a Production-Ready AI Agent API with FastAPI

The Problem with Most AI Agent APIs

Most tutorials stop at:

@app.post("/chat")
def chat(message: str):
    response = agent.run(message)
    return {"response": response}

This works for a demo. In production, you'll hit these problems immediately:

• No streaming — Users stare at a blank screen for 10+ seconds waiting for the full response
• No session management — Every request is stateless; the agent forgets everything between calls
• No rate limiting — One user can spam your API and drain your API credits
• No error handling — When the LLM times out, your whole API crashes
• No observability — You have no idea what's happening, what's failing, or what's costing money
• No tool sandboxing — Your agent can execute arbitrary code with no guardrails

Architecture Overview

┌─────────────┐     ┌──────────────┐     ┌─────────────────┐
│   Client     │────▶│  FastAPI     │────▶│  Agent Engine   │
│  (HTTP/SSE)  │◀────│  Gateway     │◀────│  (Orchestrator) │
└─────────────┘     └──────┬───────┘     └────────┬────────┘
                           │                       │
                    ┌──────▼───────┐     ┌────────▼────────┐
                    │  Rate Limiter │     │  Tool Registry  │
                    │  Session Mgr  │     │  (Sandboxed)    │
                    └──────────────┘     └─────────────────┘
                           │
                    ┌──────▼───────┐
                    │  Observability│
                    │  (Logs/Metrics)│
                    └──────────────┘

Step 1: Project Structure

ai-agent-api/
├── app/
│   ├── main.py              # FastAPI entry point
│   ├── config.py            # Settings & env vars
│   ├── models/
│   │   ├── requests.py      # Request schemas
│   │   └── responses.py     # Response schemas
│   ├── agents/
│   │   ├── engine.py        # Core agent logic
│   │   ├── tools.py         # Tool definitions
│   │   └── memory.py        # Session memory
│   ├── middleware/
│   │   ├── rate_limit.py    # Rate limiting
│   │   └── auth.py          # API key auth
│   └── observability/
│       └── logger.py        # Structured logging
├── tests/
│   ├── test_api.py
│   └── test_agent.py
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── .env.example

Step 2: Configuration with Pydantic Settings

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    # LLM
    openai_api_key: str = ""
    anthropic_api_key: str = ""
    default_model: str = "gpt-4o-mini"

    # API
    api_keys: str = ""  # Comma-separated valid API keys
    rate_limit_per_minute: int = 20

    # Sessions
    session_ttl_seconds: int = 3600  # 1 hour
    max_sessions: int = 1000

    # Observability
    log_level: str = "INFO"
    enable_metrics: bool = True

    class Config:
        env_file = ".env"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

Step 3: Session Memory Manager

# app/agents/memory.py
import time
import uuid
from collections import OrderedDict
from threading import Lock
from app.models.requests import ChatMessage

class SessionManager:
    """Thread-safe session store with TTL-based eviction."""

    def __init__(self, max_sessions: int = 1000, ttl_seconds: int = 3600):
        self._sessions: OrderedDict[str, dict] = OrderedDict()
        self._max_sessions = max_sessions
        self._ttl = ttl_seconds
        self._lock = Lock()

    def get_or_create(self, session_id=None):
        with self._lock:
            if session_id is None or session_id not in self._sessions:
                sid = session_id or str(uuid.uuid4())
                self._sessions[sid] = {
                    "messages": [],
                    "created_at": time.time(),
                    "last_accessed": time.time(),
                }
                self._evict_if_needed()
                return sid, []

            session = self._sessions[session_id]
            session["last_accessed"] = time.time()
            self._sessions.move_to_end(session_id)
            messages = [ChatMessage(**m) for m in session["messages"]]
            return session_id, messages

    def add_message(self, session_id: str, message: ChatMessage):
        with self._lock:
            if session_id in self._sessions:
                self._sessions[session_id]["messages"].append(
                    message.model_dump()
                )
                self._sessions[session_id]["last_accessed"] = time.time()

    def _evict_if_needed(self):
        now = time.time()
        expired = [
            sid for sid, s in self._sessions.items()
            if now - s["last_accessed"] > self._ttl
        ]
        for sid in expired:
            del self._sessions[sid]
        while len(self._sessions) > self._max_sessions:
            self._sessions.popitem(last=False)

Step 4: Tool Registry with Sandboxing

# app/agents/tools.py
import subprocess
import json
from pydantic import BaseModel

class ToolDefinition(BaseModel):
    name: str
    description: str
    parameters: dict
    handler: callable
    requires_approval: bool = False

class ToolRegistry:
    """Sandboxed tool execution with timeout and output limits."""

    def __init__(self):
        self._tools: dict[str, ToolDefinition] = {}

    def register(self, definition: ToolDefinition):
        self._tools[definition.name] = definition

    def get_openai_schema(self) -> list[dict]:
        return [
            {
                "type": "function",
                "function": {
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.parameters,
                }
            }
            for t in self._tools.values()
        ]

    async def execute(self, tool_name: str, arguments: dict) -> str:
        if tool_name not in self._tools:
            return json.dumps({"error": f"Unknown tool: {tool_name}"})

        tool = self._tools[tool_name]
        try:
            result = await tool.handler(**arguments)
            if len(str(result)) > 5000:
                result = str(result)[:5000] + "... [truncated]"
            return str(result)
        except Exception as e:
            return json.dumps({"error": str(e)})

async def execute_python(code: str) -> str:
    """Execute Python code in a sandboxed environment."""
    try:
        result = subprocess.run(
            ["python3", "-c", code],
            capture_output=True, text=True, timeout=10,
            env={"PATH": "/usr/bin"},
        )
        if result.returncode != 0:
            return f"Error: {result.stderr[:500]}"
        return result.stdout[:2000]
    except subprocess.TimeoutExpired:
        return "Error: Code execution timed out (10s limit)"

Step 5: The Agent Engine (Streaming + Tool Calling)

# app/agents/engine.py
import json
import time
from openai import AsyncOpenAI
from app.config import get_settings
from app.agents.memory import SessionManager
from app.agents.tools import create_default_registry

class AgentEngine:
    def __init__(self):
        self.settings = get_settings()
        self.client = AsyncOpenAI(api_key=self.settings.openai_api_key)
        self.sessions = SessionManager(
            max_sessions=self.settings.max_sessions,
            ttl_seconds=self.settings.session_ttl_seconds,
        )
        self.tools = create_default_registry()

    async def chat(self, message, session_id=None, model=None):
        model = model or self.settings.default_model
        sid, history = self.sessions.get_or_create(session_id)

        messages = self._build_messages(history, message)

        # First LLM call with tool support
        response = await self.client.chat.completions.create(
            model=model, messages=messages,
            tools=self.tools.get_openai_schema(),
            tool_choice="auto", stream=True,
        )

        tool_calls = []
        assistant_content = ""

        async for chunk in response:
            delta = chunk.choices[0].delta
            if delta.content:
                assistant_content += delta.content
                yield self._sse_chunk({
                    "type": "content", "content": delta.content,
                    "session_id": sid,
                })
            if delta.tool_calls:
                for tc in delta.tool_calls:
                    if tc.index >= len(tool_calls):
                        tool_calls.append({
                            "id": tc.id,
                            "name": tc.function.name if tc.function else "",
                            "arguments": tc.function.arguments if tc.function else "",
                        })
                    else:
                        if tc.function and tc.function.arguments:
                            tool_calls[tc.index]["arguments"] += tc.function.arguments

        # Execute tool calls
        if tool_calls:
            for tc in tool_calls:
                args = json.loads(tc["arguments"]) if tc["arguments"] else {}
                result = await self.tools.execute(tc["name"], args)
                yield self._sse_chunk({
                    "type": "tool_result", "tool": tc["name"],
                    "result": result[:200], "session_id": sid,
                })

            # Final response after tools
            final = await self.client.chat.completions.create(
                model=model, messages=messages, stream=True,
            )
            async for chunk in final:
                delta = chunk.choices[0].delta
                if delta.content:
                    yield self._sse_chunk({
                        "type": "content", "content": delta.content,
                        "session_id": sid,
                    })

        yield self._sse_chunk({
            "type": "done", "session_id": sid,
            "tokens_used": 0, "tools_called": [tc["name"] for tc in tool_calls],
        })

    def _sse_chunk(self, data):
        return f"data: {json.dumps(data)}\n\n"

Step 6: Rate Limiting Middleware

# app/middleware/rate_limit.py
import time
from collections import defaultdict
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, requests_per_minute: int = 20):
        super().__init__(app)
        self.rpm = requests_per_minute
        self._requests: dict[str, list[float]] = defaultdict(list)

    async def dispatch(self, request: Request, call_next):
        if request.url.path in ("/health", "/docs", "/openapi.json"):
            return await call_next(request)

        api_key = request.headers.get("X-API-Key", "anonymous")
        now = time.time()
        self._requests[api_key] = [
            t for t in self._requests[api_key] if now - t < 60
        ]

        if len(self._requests[api_key]) >= self.rpm:
            raise HTTPException(
                status_code=429,
                detail={
                    "error": "Rate limit exceeded",
                    "retry_after": int(60 - (now - self._requests[api_key][0])),
                },
            )

        self._requests[api_key].append(now)
        response = await call_next(request)
        response.headers["X-RateLimit-Remaining"] = str(
            self.rpm - len(self._requests[api_key])
        )
        return response

Step 7: The FastAPI Application

# app/main.py
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse
from app.config import get_settings
from app.agents.engine import AgentEngine
from app.middleware.rate_limit import RateLimitMiddleware

agent = None

@asynccontextmanager
async def lifespan(app):
    global agent
    agent = AgentEngine()
    yield

app = FastAPI(title="AI Agent API", version="1.0.0", lifespan=lifespan)
settings = get_settings()

app.add_middleware(RateLimitMiddleware,
    requests_per_minute=settings.rate_limit_per_minute)

@app.get("/health")
async def health():
    return {"status": "healthy",
            "active_sessions": agent.sessions.active_sessions}

@app.post("/chat")
async def chat(request, api_key: str = Header(...)):
    return StreamingResponse(
        agent.chat(request.message, request.session_id, request.model),
        media_type="text/event-stream",
    )

Step 8: Dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app/ ./app/
RUN useradd --create-home appuser
USER appuser
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0",
     "--port", "8000", "--workers", "4"]

Deploy It

# Local
uvicorn app.main:app --reload

# Docker
docker compose up --build

# Production: Push to GitHub → Railway/Render/Fly.io

What You Built

A production-ready AI agent API with:

• ✅ Streaming SSE responses — users see output in real-time
• ✅ Session memory — agents remember context across requests
• ✅ Tool calling — search, code execution, URL fetching
• ✅ Rate limiting — per-API-key token bucket
• ✅ API key auth — simple but effective
• ✅ Structured logging — know what's happening
• ✅ Docker support — deploy anywhere
• ✅ Health endpoint — monitoring-ready

This is the same architecture pattern we use at ZOO for client projects.