How to Productionize Agno Agents Using FastAPI with AgentOS

Agno's AgentOS class automatically constructs a production-ready FastAPI application that wires every agent, team, and workflow into secure HTTP endpoints with integrated lifecycle management and streaming support.

Deploying AI agents at scale requires more than simple HTTP wrappers—it demands authentication, resource lifecycle orchestration, and efficient streaming protocols. The agno-agi/agno repository provides AgentOS, a high-level operating system that productionizes Agno agents using FastAPI by transforming declarative Python objects into enterprise-grade REST services without boilerplate code.

FastAPI Application Architecture

AgentOS follows a structured three-phase initialization process to generate the ASGI application.

Application Factory Implementation

The foundation begins with AgentOS._make_app() in libs/agno/agno/os/app.py (lines 452-460), which instantiates and configures the core FastAPI instance. This method applies the OS title, version metadata, and OpenAPI configuration settings to ensure the generated API documentation accurately reflects your deployment.

Dynamic Router Injection

After creation, AgentOS._add_router() (lines 813-865) attaches specialized routers for agents, teams, workflows, health checks, and system utilities. This method handles route conflict detection and resolution, ensuring that user-defined endpoints never collide with system routes while maintaining clean URL hierarchies.

Unified Lifespan Orchestration

Production deployments require complex startup and shutdown sequences. The _combine_app_lifespans() method (lines 139-166) aggregates multiple asynchronous context managers into a single FastAPI lifespan handler. This orchestrates MCP tool connections, database initialization and teardown, scheduler startup, and httpx client cleanup in deterministic order, ensuring graceful resource management across the application lifecycle.

Agent Execution Endpoint

The core production interface resides in libs/agno/agno/os/routers/agents/router.py, where get_agent_router() constructs an APIRouter exposing the critical POST /agents/{agent_id}/runs endpoint (lines 60-88 and 79-106).

Security and Access Control

Every request passes through a multi-layered security stack generated by get_authentication_dependency(settings) (lines 68-71), which validates credentials against the OS-wide authentication configuration. Authorization enforcement occurs via require_resource_access("agents", "run", "agent_id") (lines 112-115), guaranteeing that callers possess explicit permission to execute the target agent.

Streaming and Real-Time Communication

When clients specify stream=true, the endpoint utilizes Server-Sent Events (SSE) to deliver real-time progress updates. The handler yields structured events including RunStarted, RunProgress, and RunFinished objects formatted via format_sse_event() (lines 126-149), allowing FastAPI to automatically generate compliant text/event-stream responses for long-running agent operations.

File Processing and Background Execution

The endpoint accepts multimodal inputs through UploadFile parameters (lines 122-125), automatically validating supported media types before injection into the agent's run() logic. For fire-and-forget scenarios, setting background=true (lines 124-127) delegates execution to FastAPI's background task system, returning an immediate HTTP response while the agent continues processing asynchronously.

Request State Propagation

AgentOS extracts user identity, session metadata, and request context from request.state (populated by upstream middleware in lines 29-50) and injects these values into the agent run kwargs, ensuring agents have access to authentication context and tracing information without manual parameter passing.

Production Deployment Pipeline

When AgentOS.get_app() is invoked, it executes a deterministic assembly sequence:

1. Base Application – Invokes _make_app() to create the configured FastAPI instance.
2. Lifespan Wrapping – Wraps any user-provided lifespan to receive the AgentOS instance while preserving system lifespans.
3. Resource Initialization – Sequentially adds MCP tool lifespan, database lifespan, scheduler lifespan, and httpx cleanup handlers via _combine_app_lifespans().
4. Route Registration – Attaches all built-in routers including home, health, base, agents, teams, workflows, and websockets through _add_built_in_routes().
5. Middleware Installation – Applies CORS configuration via update_cors_middleware() and optionally injects JWT validation through _add_jwt_middleware().

The resulting application exposes every registered agent at /agents/{id}/runs, handles authentication/authorization, manages CORS policies, and guarantees graceful startup and shutdown of all database connections, MCP servers, and background schedulers.

Implementation Examples

Creating a Production AgentOS Server

The following example demonstrates instantiating AgentOS with a custom echo agent and serving it via uvicorn:

# example.py

from agno.agent import Agent
from agno.os import AgentOS
import uvicorn

# Minimal echo agent implementation

class EchoAgent(Agent):
    async def run(self, message: str, **kwargs):
        return f"ECHO: {message}"

# Configure the OS with the agent

os = AgentOS(
    name="EchoOS",
    agents=[EchoAgent(name="echo")],
    enable_mcp_server=False,          # Disables the optional /mcp RPC endpoint

)

# Generate the FastAPI application

app = os.get_app()

# Production server entrypoint

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

The /agents/echo/runs endpoint now accepts POST requests containing a message parameter, optional file attachments, and a stream flag for SSE responses.

Streaming Request via cURL

curl -X POST "http://localhost:8000/agents/echo/runs" \
     -F "message=Hello world" \
     -F "stream=true"

When streaming is enabled, the response produces SSE-formatted output:

event: RunStarted
data: {"content":"ECHO: Hello world","run_id":"..."}

Injecting Custom Middleware

AgentOS exposes the underlying FastAPI instance for customization:

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class LogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print(f"Incoming {request.method} {request.url.path}")
        response = await call_next(request)
        return response

os = AgentOS(...)
app = os.get_app()
app.add_middleware(LogMiddleware)

Custom middleware executes before AgentOS routers, preserving the integrity of the production stack while allowing request logging, metrics collection, or header manipulation.

Key Source Files

File	Role
libs/agno/agno/os/app.py	Core AgentOS class implementing app creation, lifespan orchestration, and router registration
libs/agno/agno/os/routers/agents/router.py	/agents/{agent_id}/runs endpoint implementation with SSE and file upload support
libs/agno/agno/os/router.py	Base router utilities including health check and system route definitions
libs/agno/agno/os/utils.py	Helper functions for CORS configuration and route conflict detection
libs/agno/agno/os/middleware/jwt.py	Optional JWT authentication middleware activated when authorization=True
libs/agno/agno/os/auth.py	Authentication dependency generator for request validation
libs/agno/agno/os/mcp.py	Model Context Protocol server implementation for external tool integration

Summary

• AgentOS transforms declarative agent definitions into production FastAPI applications through get_app(), eliminating infrastructure boilerplate.
• The three-phase initialization (app creation, router injection, lifespan orchestration) ensures reliable resource management and clean separation of concerns.
• Agents expose a unified REST interface at /agents/{agent_id}/runs supporting synchronous responses, SSE streaming, background execution, and multipart file uploads.
• Integrated security layers handle authentication via configurable dependencies and fine-grained authorization through resource access controls.
• Automatic lifecycle management coordinates database connections, MCP tool servers, schedulers, and HTTP clients during startup and shutdown.

Frequently Asked Questions

How does AgentOS handle authentication for agent endpoints?

AgentOS generates authentication dependencies through get_authentication_dependency(settings) in libs/agno/agno/os/auth.py, which is applied to all agent routes during router creation. When requests arrive, the dependency validates credentials against the OS-wide configuration before reaching the endpoint handler, ensuring every agent execution occurs within an authenticated context.

Can I run Agno agents asynchronously in the background via the API?

Yes. The /agents/{agent_id}/runs endpoint accepts a background=true parameter that delegates execution to FastAPI's background task system. This returns an immediate HTTP 202 response while the agent continues processing, making it ideal for long-running operations where clients do not need to wait for completion.

What lifecycle components does AgentOS manage automatically?

AgentOS combines multiple lifespans through _combine_app_lifespans() to manage MCP tool server connections, database initialization and connection pooling, job scheduler startup/shutdown, and httpx async client cleanup. These run sequentially during application startup and reverse during shutdown to prevent resource leaks.

How do I add custom middleware to an AgentOS FastAPI application?

After calling os.get_app(), the returned object is a standard FastAPI instance. You can attach custom middleware using app.add_middleware() with Starlette middleware classes. These execute before AgentOS internal routers, allowing you to implement logging, metrics, or custom header injection without modifying the core OS code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}