AgentOS Overview

​

What is AgentOS?

• RESTful API endpoints for all agents, teams, and workflows
• WebSocket support for real-time streaming responses
• Authentication & authorization with JWT and RBAC
• Session management with automatic memory persistence
• Knowledge base integration with vector search
• Monitoring & observability with traces, metrics, and evaluations
• Scheduler for cron-based agent execution
• Approval workflows for human-in-the-loop operations

​

Key Features

​

Architecture

┌─────────────────────────────────────────┐
│   Interfaces (REST, WebSocket, Slack)   │
├─────────────────────────────────────────┤
│   AgentOS Core (Routing & Middleware)   │
├─────────────────────────────────────────┤
│   Agents, Teams, Workflows              │
├─────────────────────────────────────────┤
│   Storage (PostgreSQL, SQLite, etc.)    │
└─────────────────────────────────────────┘

​

Core Components

• Central orchestrator that manages all agents, teams, and workflows
• Handles database initialization and connection pooling
• Configures middleware, authentication, and routing
• Located at: libs/agno/agno/os/app.py:180

• /agents - Agent management and execution
• /teams - Team collaboration endpoints
• /workflows - Workflow orchestration
• /sessions - Session and conversation history
• /memory - Long-term memory management
• /knowledge - Knowledge base operations
• /schedules - Cron scheduler management
• /approvals - Human approval workflows

• REST API (default)
• WebSocket for streaming
• A2A (Agent-to-Agent protocol)
• Slack integration
• Custom interfaces via BaseInterface

​

Basic Usage

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.os import AgentOS

# Setup database
db = PostgresDb(
    id="production-db",
    db_url="postgresql+psycopg://user:pass@localhost:5432/agno"
)

# Create agent
agent = Agent(
    name="Assistant",
    db=db,
    enable_session_summaries=True,
    update_memory_on_run=True,
)

# Create AgentOS
agent_os = AgentOS(
    id="my-agent-os",
    description="Production AI Assistant",
    agents=[agent],
)

# Get FastAPI app
app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app="main:app", port=7777)

​

Key Concepts

​

Automatic Database Discovery

# Each component can have its own database
agent1 = Agent(db=postgres_db)
agent2 = Agent(db=sqlite_db)
team = Team(db=postgres_db)

# AgentOS discovers and manages all databases
agent_os = AgentOS(agents=[agent1, agent2], teams=[team])
# Auto-provisions tables and connection pools

​

Unified Configuration

agent_os = AgentOS(
    db=db,  # Default database for all agents/teams
    agents=[agent1, agent2],  # Agents without db use OS db
    auto_provision_dbs=True,  # Auto-create tables
    tracing=True,  # Enable tracing for all
    run_hooks_in_background=True,  # Non-blocking hooks
)

​

Lifespan Management

• Startup: Database initialization, MCP tool connections, scheduler startup
• Runtime: Request handling, connection pooling, background tasks
• Shutdown: Graceful database closure, connection cleanup, task completion

from contextlib import asynccontextmanager

@asynccontextmanager
async def custom_lifespan(app, agent_os):
    # Startup logic
    print(f"Starting {agent_os.id}")
    yield
    # Shutdown logic
    print("Shutting down gracefully")

agent_os = AgentOS(
    agents=[agent],
    lifespan=custom_lifespan,
)

​

Available Endpoints

​

Production Considerations

​

Performance

• Async by default: All database operations are async-first
• Connection pooling: Automatic database connection management
• Background tasks: Non-blocking hooks and evaluations
• Streaming responses: Memory-efficient SSE streaming

​

Scalability

• Horizontal scaling: Stateless design for multi-instance deployment
• Database sharding: Multiple database support per component
• Resource isolation: Separate databases per agent/team

​

Reliability

• Health checks: Built-in /health endpoint
• Error handling: Comprehensive exception handling
• Graceful shutdown: Clean resource cleanup
• Retry logic: Automatic retry for transient failures

​

Next Steps