# How to Productionize Agno Agents Using FastAPI with AgentOS

> Learn how Agno's AgentOS productionizes agents with FastAPI. Automatically create secure HTTP endpoints for agents, teams, and workflows with lifecycle management and streaming.

- Repository: [Agno/agno](https://github.com/agno-agi/agno)
- Tags: how-to-guide
- Published: 2026-02-23

---

**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`](https://github.com/agno-agi/agno/blob/main/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`](https://github.com/agno-agi/agno/blob/main/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:

```python

# 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

```bash
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:

```python
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`](https://github.com/agno-agi/agno/blob/main/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`](https://github.com/agno-agi/agno/blob/main/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`](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/os/router.py) | Base router utilities including health check and system route definitions |
| [`libs/agno/agno/os/utils.py`](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/os/utils.py) | Helper functions for CORS configuration and route conflict detection |
| [`libs/agno/agno/os/middleware/jwt.py`](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/os/middleware/jwt.py) | Optional JWT authentication middleware activated when `authorization=True` |
| [`libs/agno/agno/os/auth.py`](https://github.com/agno-agi/agno/blob/main/libs/agno/agno/os/auth.py) | Authentication dependency generator for request validation |
| [`libs/agno/agno/os/mcp.py`](https://github.com/agno-agi/agno/blob/main/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`](https://github.com/agno-agi/agno/blob/main/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.