# AgentOS Authentication Source: https://docs.agno.com/agent-os/api/authentication Authenticate with AgentOS using RBAC and JWT tokens AgentOS uses Role-Based Access Control (RBAC) with JWT tokens to secure your API endpoints and provide fine-grained permission control. ## RBAC Authentication (Recommended) When RBAC is enabled, include a JWT token with scopes in your requests: ```bash theme={null} curl --location 'http://localhost:7777/agents/my-agent/runs' \ --header 'Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...' \ --header 'Content-Type: application/x-www-form-urlencoded' \ --data-urlencode 'message=Your query here' ``` ### JWT Token Structure Your JWT tokens should include scopes and audience claims: ```json theme={null} { "sub": "user-123", "aud": "my-agent-os", "scopes": ["agents:read", "agents:my-agent:run", "sessions:write"], "exp": 1735689600 } ``` ### Required Claims | Claim | Description | | -------- | ---------------------------- | | `aud` | Must match your AgentOS `id` | | `scopes` | Array of permission scopes | ### Common Scopes | Scope | Description | | ----------------- | ---------------------- | | `agents:read` | List and view agents | | `agents:run` | Run any agent | | `agents::run` | Run a specific agent | | `sessions:read` | View session data | | `sessions:write` | Create/update sessions | | `agent_os:admin` | Full admin access | See [RBAC Documentation](/agent-os/security/rbac) for all available scopes. ### Error Responses | Status | Description | | ------------------ | ----------------------------------- | | `401 Unauthorized` | Missing or invalid JWT token | | `401 Unauthorized` | Invalid audience (wrong AgentOS ID) | | `403 Forbidden` | Insufficient scopes | ## Developer Resources Enable RBAC and configure authorization. Complete scope reference and endpoint mappings. JWT configuration and parameter injection. *** ## Security Key Authentication (Deprecated) Security Key authentication is deprecated. Use [RBAC](#rbac-authentication-recommended) instead. Legacy authentication using the `OS_SECURITY_KEY` environment variable: ```bash theme={null} curl --location 'http://localhost:7777/agents/my-agent/runs' \ --header 'Authorization: Bearer your-security-key' \ --header 'Content-Type: application/x-www-form-urlencoded' \ --data-urlencode 'message=Your query here' ``` # API Usage Source: https://docs.agno.com/agent-os/api/usage Running your Agent / Team / Workflow with the AgentOS API The AgentOS API provides endpoints: * **Run an Agent**: `POST /agents/{agent_id}/runs` (See the [API reference](/reference-api/schema/agents/create-agent-run)) * **Run a Team**: `POST /teams/{team_id}/runs` (See the [API reference](/reference-api/schema/teams/create-team-run)) * **Run a Workflow**: `POST /workflows/{workflow_id}/runs` (See the [API reference](/reference-api/schema/workflows/execute-workflow)) These endpoints support form-based input. Below is an example of how to run an agent with the API: ```bash theme={null} curl --location 'http://localhost:7777/agents/agno-agent/runs' \ --header 'Content-Type: application/x-www-form-urlencoded' \ --data-urlencode 'message=Tell me about Agno.' \ --data-urlencode 'stream=True' \ --data-urlencode 'user_id=john@example.com' \ --data-urlencode 'session_id=session_123' ``` ## Passing parameters to your Agent / Team / Workflow Agent, Team and Workflow `run()` and `arun()` endpoints support various runtime parameters. See the [Agent run schema](/reference/agents/agent#run), [Team run schema](/reference/teams/team#run), [Workflow run schema](/reference/workflows/workflow#run) for more details. It is a common pattern to want to pass `session_state`, `dependencies`, `metadata`, etc. to your Agent, Team or Workflow via the API. To pass these parameters via the AgentOS API, you can simply specify them as form-based parameters. Below is an example where `dependencies` are passed to the agent: ```python dependencies_to_agent.py theme={null} from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.os import AgentOS # Setup the database db = PostgresDb(id="basic-db", db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") # Setup basic agents, teams and workflows story_writer = Agent( id="story-writer-agent", name="Story Writer Agent", db=db, markdown=True, instructions="You are a story writer. You are asked to write a story about a robot. Always name the robot {robot_name}", ) # Setup our AgentOS app agent_os = AgentOS( description="Example AgentOS to show how to pass dependencies to an agent", agents=[story_writer], ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="dependencies_to_agent:app", reload=True) ``` Then to test it, you can run the following command: ```bash theme={null} curl --location 'http://localhost:7777/agents/story-writer-agent/runs' \ --header 'Content-Type: application/x-www-form-urlencoded' \ --data-urlencode 'message=Write me a 5 line story.' \ --data-urlencode 'dependencies={"robot_name": "Anna"}' ``` ## Passing Output Schema You can pass an output schema for a specific agent or team run by passing the `output_schema` parameter as a JSON schema string. By default, the schema is converted to a Pydantic model. ```bash theme={null} curl --location 'http://localhost:7777/agents/story-writer-agent/runs' \ --header 'Content-Type: application/x-www-form-urlencoded' \ --data-urlencode 'message=Write a story' \ --data-urlencode 'output_schema={"type":"object","properties":{"title":{"type":"string"},"content":{"type":"string"}},"required":["title","content"]}' ``` To keep the output schema as a JSON dict instead of converting to a Pydantic model, set `use_json_schema=true`: ```bash theme={null} curl --location 'http://localhost:7777/agents/story-writer-agent/runs' \ --header 'Content-Type: application/x-www-form-urlencoded' \ --data-urlencode 'message=Write a story' \ --data-urlencode 'output_schema={"type":"json_schema","json_schema":{"name":"StoryOutput","schema":{"type":"object","properties":{"title":{"type":"string"},"content":{"type":"string"}},"required":["title","content"],"additionalProperties":false}}}' \ --data-urlencode 'use_json_schema=true' ``` ## Cancelling a Run You can cancel a running agent, team or workflow by using the appropriate endpoint. For example, to cancel an agent run: ```bash theme={null} curl --location 'http://localhost:7777/agents/story-writer-agent/runs/123/cancel' ``` See the specific API references for more details: * [Cancel Agent Run API reference](/reference-api/schema/agents/cancel-agent-run) * [Cancel Team Run API reference](/reference-api/schema/teams/cancel-team-run) * [Cancel Workflow Run API reference](/reference-api/schema/workflows/cancel-workflow-run) ## Developer Resources * View [Complete API Reference](/reference-api/overview) for all endpoints * See [Authentication](/agent-os/api/authentication) for securing your API # Background Hooks Source: https://docs.agno.com/agent-os/background-tasks/overview Run agent hooks as non-blocking background tasks in AgentOS When serving agents or teams through AgentOS, you can configure pre-hooks and post-hooks to run as background tasks. This means the API response is returned immediately to the user while the hooks continue executing in the background. ## Why Use Background Hooks? By default, hooks used by agents and teams in your AgentOS are in the execution path and block the response: Background tasks not enabled Background tasks not enabled With background hooks enabled, your hooks won't block the response, increasing response speed: Background tasks enabled Background tasks enabled This is useful for: * **Agent Evaluation**: Evaluate the agent's responses without affecting the responses themselves * **Analytics and logging**: Track usage patterns without affecting response time * **Notifications**: Send emails, Slack messages, or webhook calls * **External API calls**: Sync data with third-party services * **Non-critical data processing**: Tasks that don't affect the response ## Enabling Background Tasks There are two ways to enable background execution for hooks: ### Option 1: Global Setting via AgentOS Enable background execution for **all** hooks across all agents and teams (even as part of a workflow): ```python theme={null} from agno.os import AgentOS agent_os = AgentOS( agents=[agent], teams=[team], workflows=[workflow], run_hooks_in_background=True, # All hooks run in background ) ``` When enabled, this setting automatically propagates to: * All agents registered with AgentOS * All teams and their member agents (including nested teams) * All workflows and the agents/teams within their steps See [Global Background Hooks Example](/agent-os/usage/background-hooks-global) for an example. Note that pre-hooks are typically used for validation or modification of the input of a run. If you use them as background tasks, they will execute after the run has already been initiated. If you have hooks that should not run as background tasks, you should use the second option and mark only the specific hooks to run in background. ### Option 2: Per-Hook Setting via Decorator Mark specific hooks to run in background using the `@hook` decorator: ```python theme={null} from agno.hooks import hook @hook(run_in_background=True) async def send_notification(run_output, agent): """Only this hook runs in the background.""" await send_slack_message(run_output.content) ``` This approach gives you fine-grained control: critical hooks are executed during the run while non-critical hooks run in the background. See [Per-Hook Background Example](/agent-os/usage/background-hooks-decorator) for an example. **Background tasks require AgentOS.** When running agents directly (not through AgentOS), the `@hook(run_in_background=True)` decorator has no effect - hooks will run synchronously. ## How It Works AgentOS uses FastAPI's [BackgroundTasks](https://fastapi.tiangolo.com/tutorial/background-tasks/) to schedule hooks for execution after the response is sent. Background tasks execute **sequentially** after the response is sent. If you have multiple background hooks, they run one after another. **Pre- and post-hooks in background mode cannot modify the request or response.** Any modifications to `run_input` or `run_output` won't affect the agent's processing. Only use background mode for pre- and post-hooks that perform logging or monitoring. This means background mode is not suitable for Guardrails. ### Data Isolation When hooks run in the background, AgentOS automatically creates deep copies of: * `run_input` - The input to the agent run * `run_context` - The current run context * `run_output` - The output from the agent This prevents race conditions where background hooks might accidentally modify data that's being used elsewhere. ### Error Handling Errors in background tasks don't affect the API response (since it's already been sent). Make sure to implement proper error handling and logging in your background hooks: ```python theme={null} @hook(run_in_background=True) async def safe_background_hook(run_output, agent): try: await external_api_call(run_output) except Exception as e: logger.error(f"Background hook failed: {e}") ``` ## Examples Enable background hooks globally for all agents Mix synchronous and background hooks Use an agent-as-judge to evaluate responses ## Developer Resources * View [@hook Decorator Reference](/reference/hooks/hook-decorator) * View [Hooks Overview](/basics/hooks/overview) * View [Cookbook Examples](https://github.com/agno-agi/agno/tree/main/cookbook/06_agent_os/background_tasks) # AgentOS Client Source: https://docs.agno.com/agent-os/client/overview Python client for programmatic access to AgentOS instances The `AgentOSClient` provides a Python interface for interacting with running AgentOS instances. It enables you to: * **Run agents, teams, and workflows** programmatically with streaming support * **Manage sessions** for conversation persistence across runs * **Search and manage knowledge** in connected knowledge bases * **Access memories** stored for users * **Monitor traces** for debugging and observability ## Quick Start ```python theme={null} import asyncio from agno.client import AgentOSClient async def main(): # Connect to AgentOS client = AgentOSClient(base_url="http://localhost:7777") # Get configuration and available agents config = await client.aget_config() print(f"Connected to: {config.name or config.os_id}") print(f"Available agents: {[a.id for a in config.agents]}") # Run an agent if config.agents: result = await client.run_agent( agent_id=config.agents[0].id, message="Hello, how can you help me?", ) print(f"Response: {result.content}") asyncio.run(main()) ``` ## Streaming Responses The client supports streaming for real-time response handling: ```python theme={null} from agno.run.agent import RunContentEvent, RunCompletedEvent async for event in client.run_agent_stream( agent_id="my-agent", message="Tell me a story", ): if isinstance(event, RunContentEvent): print(event.content, end="", flush=True) elif isinstance(event, RunCompletedEvent): print(f"\nCompleted! Run ID: {event.run_id}") ``` ## Authentication When connecting to authenticated AgentOS instances, pass headers with your requests: ```python theme={null} headers = {"Authorization": "Bearer your-jwt-token"} config = await client.aget_config(headers=headers) result = await client.run_agent( agent_id="my-agent", message="Hello", headers=headers, ) ``` ## Error Handling ```python theme={null} from agno.exceptions import RemoteServerUnavailableError try: config = await client.aget_config() except RemoteServerUnavailableError as e: print(f"Server unavailable: {e.message}") print(f"URL: {e.base_url}") ``` ## Examples Connect and explore an AgentOS instance Execute streaming and non-streaming agent runs Execute team runs with member coordination Execute workflow pipelines Create, list, and manage sessions Search and upload knowledge content Create, list, and manage user memories ## API Reference For complete method documentation and parameters, see the [AgentOSClient Reference](/reference/agent-os/client). # AgentOS Configuration Source: https://docs.agno.com/agent-os/config Customize your AgentOS instance with custom configuration AgentOS configuration allows you to customize your instance for different environments and deployment scenarios. You can control which AI models are available globally and per-domain (for Evals), set custom display names for UI pages, define "quick prompts" for the chat interface, and configure per-database settings. This is particularly useful when managing multiple databases, deploying across different environments (development, staging, production), or building multi-tenant systems where each client needs distinct configurations. Configuring "Quick Prompts" can be particularly useful for improving the chat experience for users of your AgentOS. It changes the available options when your user creates a new session on the Chat Page. ## Setting your configuration You can provide your AgentOS configuration in two different ways: with a configuration YAML file, or using the `AgentOSConfig` class. See the full reference for the `AgentOSConfig` class [here](/reference/agent-os/configuration). ## Configuration YAML File 1. Create a YAML file with your configuration. For example: ```yaml theme={null} # Configure quick prompts for the Chat interface (per agent) chat: quick_prompts: marketing-agent: - "What can you do?" - "How is our latest post working?" - "Tell me about our active marketing campaigns" # Configure Memory page with custom display names memory: display_name: "User Memory Store" dbs: - db_id: db-0001 tables: ["custom_memory_table"] # Optional: specify custom table names domain_config: display_name: Main app user memories - db_id: db-0002 domain_config: display_name: Support flow user memories # Configure Knowledge page knowledge: display_name: "Knowledge Base" dbs: - db_id: db-0001 domain_config: display_name: Product documentation # Configure Session tracking session: display_name: "User Sessions" dbs: - db_id: db-0001 domain_config: display_name: Production sessions # Configure Evals page (supports both global and per-database available models) evals: display_name: "Evaluations" available_models: - "openai:gpt-4" - "anthropic:claude-sonnet-4" dbs: - db_id: db-0001 domain_config: display_name: Production evals available_models: - "openai:gpt-4o-mini" ``` 2. Pass the configuration to your AgentOS using the `config` parameter: ```python theme={null} from agno.os import AgentOS agent_os = AgentOS( name="My AgentOS", ..., config="path/to/configuration.yaml" ) ``` ## `AgentOSConfig` Class You can also provide your configuration using the `AgentOSConfig` class: ```python theme={null} from agno.os import AgentOS from agno.os.config import ( AgentOSConfig, ChatConfig, DatabaseConfig, EvalsConfig, EvalsDomainConfig, KnowledgeConfig, KnowledgeDomainConfig, MemoryConfig, MemoryDomainConfig, SessionConfig, SessionDomainConfig, ) agent_os = AgentOS( ..., config=AgentOSConfig( chat=ChatConfig( quick_prompts={ "marketing-agent": [ "What can you do?", "How is our latest post working?", "Tell me about our active marketing campaigns", ] } ), memory=MemoryConfig( display_name="User Memory Store", dbs=[ DatabaseConfig( db_id=marketing_db.id, domain_config=MemoryDomainConfig( display_name="Main app user memories", ), ), DatabaseConfig( db_id=support_db.id, domain_config=MemoryDomainConfig( display_name="Support flow user memories", ), ) ], ), ), ) ``` ## The /config endpoint The `/config` endpoint returns your complete AgentOS configuration as JSON. You could use this to inspect your AgentOS configuration that is served to the AgentOS Control Plane. The response includes: * **OS ID**: The ID of your AgentOS (automatically generated if not set) * **Description**: The description of your AgentOS * **Databases**: The list of IDs of the databases present in your AgentOS * **Agents**: The list of Agents available in your AgentOS * **Teams**: The list of Teams available in your AgentOS * **Workflows**: The list of Workflows available in your AgentOS * **Interfaces**: The list of Interfaces available in your AgentOS. E.g. WhatsApp, Slack, etc. * **Chat**: The configuration for the Chat page, which includes the list of quick prompts for each Agent, Team and Workflow in your AgentOS * **Session**: The configuration for the Session page of your AgentOS * **Metrics**: The configuration for the Metrics page of your AgentOS * **Memory**: The configuration for the Memory page of your AgentOS * **Knowledge**: The configuration for the Knowledge page of your AgentOS * **Evals**: The configuration for the Evals page of your AgentOS You will receive a JSON response with your configuration. Using the previous examples, you will receive: ```json theme={null} { "os_id": "0001", "name": "My AgentOS", "description": "Your AgentOS", "available_models": [ "openai:gpt-4", ], "databases": [ "db-0001", "db-0002" ], "agents": [], "chat": { "quick_prompts": { "marketing-agent": [ "What can you do?", "How is our latest post working?", "Tell me about our active marketing campaigns" ] } }, "memory": { "dbs": [ { "db_id": "db-0001", "domain_config": { "display_name": "Main app user memories" } }, { "db_id": "db-0002", "domain_config": { "display_name": "Support flow user memories" } } ] }, ... } ``` See the full schema for the `/config` endpoint [here](/reference-api/schema/core/get-os-configuration). # Connecting Your AgentOS Source: https://docs.agno.com/agent-os/connecting-your-os Step-by-step guide to connect your local AgentOS to the AgentOS Control Plane ## Overview Connecting your AgentOS is the critical first step to using the AgentOS Control Plane. This process establishes a connection between your running AgentOS instance and the Control Plane, allowing you to manage, monitor, and interact with your agents through the browser. **Prerequisites**: You need a running AgentOS instance before you can connect it to the Control Plane. If you haven't created one yet, check out our [Creating Your First OS](/agent-os/creating-your-first-os) guide. See the [AgentOS Control Plane](/agent-os/control-plane) documentation for more information about the Control Plane. ## Step-by-Step Connection Process ### 1. Access the Connection Dialog In the Agno platform: 1. Click on the team/organization dropdown in the top navigation bar 2. Click the **"+"** (plus) button next to "Add new OS" 3. The "Connect your AgentOS" dialog will open ### 2. Choose Your Environment Select **"Local"** for development or **"Live"** for production: * **Local**: Connects to an AgentOS running on your local machine * **Live**: Connects to a production AgentOS running on your infrastructure Live AgentOS connections require a PRO subscription. ### 3. Configure Connection Settings #### Endpoint URL * **Default Local**: `http://localhost:7777` * **Custom Local**: You can change the port if your AgentOS runs on a different port * **Live**: Enter your production HTTPS URL Make sure your AgentOS is actually running on the specified endpoint before attempting to connect. #### OS Name Give your AgentOS a descriptive name: * Use clear, descriptive names like "Development OS" or "Production Chat Bot" * This name will appear in your OS list and help you identify different instances #### Tags (Optional) Add tags to organize your AgentOS instances: * Examples: `development`, `production`, `chatbot`, `research` * Tags help filter and organize multiple OS instances * Click the **"+"** button to add multiple tags ### 4. Test and Connect 1. Click the **"CONNECT"** button 2. The platform will attempt to establish a connection to your AgentOS 3. If successful, you'll see your new OS in the organization dashboard ## Verifying Your Connection Once connected, you should see: 1. **OS Status**: "Running" indicator in the platform 2. **Available Features**: Chat, Knowledge, Memory, Sessions, etc. should be accessible 3. **Agent List**: Your configured agents should appear in the chat interface ## Securing Your Connection Protect your AgentOS APIs and Control Plane access with bearer-token authentication. Security keys provide essential protection for both development and production environments. **Key Features:** * Generate unique security keys per AgentOS instance * Rotate keys easily through the organization settings * Configure bearer-token authentication on your server For complete security setup instructions, including environment configuration and best practices, see the [Security Key](/agent-os/security) documentation. ## Managing Connected OS Instances ### Switching Between OS Instances 1. Use the dropdown in the top navigation bar 2. Select the OS instance you want to work with 3. All platform features will now connect to the selected OS ### Disconnecting an OS 1. Go to the organization settings 2. Find the OS in your list 3. Click the delete option Disconnecting an OS doesn't stop the AgentOS instance - it only removes it from the platform interface. ## Next Steps Once your AgentOS is successfully connected: Start having conversations with your connected agents Upload and organize your knowledge bases Track and analyze your agent interactions # Control Plane Source: https://docs.agno.com/agent-os/control-plane The main web interface for interacting with and managing your AgentOS instances The AgentOS Control Plane is your primary web interface for accessing and managing all AgentOS features. This intuitive dashboard serves as the central hub where you interact with your agents, manage knowledge bases, track sessions, monitor performance, and control user access. AgentOS Control Plane Dashboard ## OS Management Connect and inspect your OS runtimes from a single interface. Switch between local development and live production instances, monitor connection health, and configure endpoints for your different environments. ## User Management Manage your organization members and their access to AgentOS features. Configure your organization name, invite team members, and control permissions from a centralized interface. ### Inviting Members Add new team members to your organization by entering their email addresses. You can invite multiple users at once by separating emails with commas or pressing Enter/Tab between addresses. ### Member Roles Control what each member can access: * **Owner**: Full administrative access including billing and member management * **Member**: Access to AgentOS features and collaboration capabilities ## General Settings Configure your account preferences and organization settings. Access your profile information, manage billing and subscription details, and adjust organization-wide preferences from a centralized settings interface. ## Feature Access The control plane provides direct access to all main AgentOS capabilities through an intuitive interface: **Getting Started Tip**: The control plane is your gateway to all AgentOS features. Start by connecting your OS instance, then explore each feature section to familiarize yourself with the interface. Start conversations with your agents and access multi-agent interactions Upload and organize documents with search and browsing capabilities Browse stored memories and search through conversation history Track and analyze agent interactions and performance Test and evaluate agent performance with comprehensive metrics Monitor system performance and usage analytics ## Next Steps Ready to get started with the AgentOS control plane? Here's what you need to do: Set up a new AgentOS instance from scratch using our templates Learn how to connect your local development environment to the platform # Create Your First AgentOS Source: https://docs.agno.com/agent-os/creating-your-first-os Quick setup guide to get your first AgentOS instance running locally Get started with AgentOS by setting up a minimal local instance. This guide will have you running your first agent in minutes, with optional paths to add advanced features through our examples. AgentOS is a FastAPI app that you can run locally or in your cloud. If you want to build AgentOS using an existing FastAPI app, check out the [Custom FastAPI App](/agent-os/custom-fastapi/overview) guide. ## Prerequisites * Python 3.9+ * An LLM provider API key (e.g., `OPENAI_API_KEY`) ## Installation Create and activate a virtual environment: ```bash Mac theme={null} # Create virtual environment python -m venv venv # Activate virtual environment source venv/bin/activate ``` ```bash Windows theme={null} # Create virtual environment python -m venv venv # Activate virtual environment venv\Scripts\activate ``` Install dependencies: ```bash theme={null} pip install -U agno "fastapi[standard]" uvicorn openai ``` ## Minimal Setup Create `my_os.py`: ```python theme={null} from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.db.sqlite import AsyncSqliteDb from agno.os import AgentOS assistant = Agent( name="Assistant", model=OpenAIChat(id="gpt-5-mini"), db=AsyncSqliteDb(db_file="my_os.db"), instructions=["You are a helpful AI assistant."], markdown=True, ) agent_os = AgentOS( id="my-first-os", description="My first AgentOS", agents=[assistant], ) app = agent_os.get_app() if __name__ == "__main__": # Default port is 7777; change with port=... agent_os.serve(app="my_os:app", reload=True) ``` It is recommended to use all of the `async` building blocks of Agno when building your AgentOS. For example, use async tool definitions, async database connections, async pre/post hooks, etc. This helps unlock maximum concurrency and performance when running your AgentOS. ## Running Your OS Start your AgentOS: ```bash theme={null} python my_os.py ``` Access your running instance: * **App Interface**: `http://localhost:7777` - Use this URL when connecting to the AgentOS control plane * **API Documentation**: `http://localhost:7777/docs` - Interactive API documentation and testing * **Configuration**: `http://localhost:7777/config` - View AgentOS configuration * **API Reference**: View the [AgentOS API documentation](/reference-api/overview) for programmatic access ## Connecting to the Control Plane With your AgentOS now running locally (`http://localhost:7777`), you can connect it to the AgentOS control plane for a enhanced management experience. The control plane provides a centralized interface to interact with your agents, manage knowledge bases, track sessions, and monitor performance. ## Next Steps Connect your running OS to the AgentOS control plane interface Explore comprehensive examples for advanced AgentOS configurations # Overriding Routes Source: https://docs.agno.com/agent-os/custom-fastapi/override-routes Learn how to override AgentOS routes with your own custom routes when conflicts occur When integrating your custom FastAPI application with AgentOS, route conflicts can occur if both your app and AgentOS define the same endpoint paths. For example, both might define a `/health` endpoint or a root `/` route. AgentOS provides the `on_route_conflict` parameter to control how these conflicts are resolved, allowing you to choose whether your custom routes or AgentOS routes take precedence. ## When to Use Override routes when you need: * **Custom health checks**: Replace AgentOS's `/health` endpoint with your own monitoring logic * **Branded landing pages**: Serve a custom homepage at `/` instead of the default AgentOS interface * **Custom authentication**: Implement your own auth endpoints that conflict with AgentOS defaults * **API versioning**: Control which version of an endpoint is exposed * **Custom error handlers**: Define specialized error handling for specific routes ## Configuration Options AgentOS provides two values for the `on_route_conflict` parameter for handling route conflicts: | Option | Custom Routes | AgentOS Routes | Warnings Logged | | ---------------------------- | ------------- | -------------------------- | --------------- | | `preserve_base_app` | ✓ Preserved | ✗ Skipped (conflicts only) | Yes | | `preserve_agentos` (default) | ✗ Overridden | ✓ Preserved | Yes | Non-conflicting routes from both your app and AgentOS are always included regardless of the mode. ## Example This example demonstrates using `on_route_conflict="preserve_base_app"` to preserve custom routes for the home page and health endpoint. ```python override_routes.py theme={null} from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.anthropic import Claude from agno.os import AgentOS from agno.tools.duckduckgo import DuckDuckGoTools from fastapi import FastAPI # Set up the database db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") web_research_agent = Agent( id="web-research-agent", name="Web Research Agent", model=Claude(id="claude-sonnet-4-0"), db=db, tools=[DuckDuckGoTools()], add_history_to_context=True, num_history_runs=3, add_datetime_to_context=True, markdown=True, ) # Create custom FastAPI app app: FastAPI = FastAPI( title="Custom FastAPI App", version="1.0.0", ) # Custom landing page (conflicts with AgentOS home route) @app.get("/") async def get_custom_home(): return { "message": "Custom FastAPI App", "note": "Using on_route_conflict=\"preserve_base_app\" to preserve custom routes", } # Custom health endpoint (conflicts with AgentOS health route) @app.get("/health") async def get_custom_health(): return {"status": "custom_ok", "note": "This is your custom health endpoint"} # Set up the AgentOS app by passing your FastAPI app # Use on_route_conflict="preserve_base_app" to preserve your custom routes over AgentOS routes agent_os = AgentOS( description="Example app with route replacement", agents=[web_research_agent], base_app=app, on_route_conflict="preserve_base_app", # Skip conflicting AgentOS routes, keep your custom routes ) app = agent_os.get_app() if __name__ == "__main__": """Run the AgentOS application. With on_route_conflict="preserve_base_app": - Your custom routes are preserved: http://localhost:7777/ and http://localhost:7777/health - AgentOS routes are available at other paths: http://localhost:7777/sessions, etc. - Conflicting AgentOS routes (GET / and GET /health) are skipped - API docs: http://localhost:7777/docs Try changing on_route_conflict to "preserve_agentos" to see AgentOS routes override your custom ones. """ agent_os.serve(app="override_routes:app", reload=True) ``` # Bring Your Own FastAPI App Source: https://docs.agno.com/agent-os/custom-fastapi/overview Learn how to use your own FastAPI app in your AgentOS AgentOS is built on FastAPI, which means you can easily integrate your existing FastAPI applications or add custom routes and routers to extend your agent's capabilities. ## Quick Start The simplest way to bring your own FastAPI app is to pass it to the AgentOS constructor: ```python theme={null} from fastapi import FastAPI from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.os import AgentOS # Create your custom FastAPI app app = FastAPI(title="My Custom App") # Add your custom routes @app.get("/status") async def status_check(): return {"status": "healthy"} # Pass your app to AgentOS agent_os = AgentOS( agents=[Agent(id="basic-agent", model=OpenAIChat(id="gpt-5-mini"))], base_app=app # Your custom FastAPI app ) # Get the combined app with both AgentOS and your routes app = agent_os.get_app() ``` Your custom FastAPI app can have its own middleware and dependencies. If you have your own CORS middleware, it will be updated to include the AgentOS allowed origins, to make the AgentOS instance compatible with the Control Plane. Otherwise, the appropriate CORS middleware will be added to the app. ### Adding Middleware You can add any FastAPI middleware to your custom FastAPI app and it will be respected by AgentOS. Agno also provides some built-in middleware for common use cases, including authentication. See the [Middleware](/agent-os/middleware/overview) page for more details. ### Running with FastAPI CLI AgentOS applications are compatible with the [FastAPI CLI](https://fastapi.tiangolo.com/deployment/manually/) for development. First, install the FastAPI CLI: ```bash Install FastAPI CLI theme={null} pip install "fastapi[standard]" ``` Then run the app: ```bash Run with FastAPI CLI theme={null} fastapi run your_app.py ``` ```bash Run with auto-reload theme={null} fastapi run your_app.py --reload ``` ```bash Custom host and port theme={null} fastapi run your_app.py --host 0.0.0.0 --port 8000 ``` ### Running in Production For production deployments, you can use any ASGI server: ```bash Uvicorn theme={null} uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 ``` ```bash Gunicorn with Uvicorn workers theme={null} gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000 ``` ```bash FastAPI CLI (Production) theme={null} fastapi run main.py --host 0.0.0.0 --port 8000 ``` ## Adding Custom Routers For better organization, use FastAPI routers to group related endpoints: ```python custom_fastapi_app.py theme={null} from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.models.anthropic import Claude from agno.os import AgentOS from agno.tools.duckduckgo import DuckDuckGoTools from fastapi import FastAPI # Set up the database db = SqliteDb(db_file="tmp/agentos.db") # Set up the agent web_research_agent = Agent( name="Basic Agent", model=Claude(id="claude-sonnet-4-0"), db=db, tools=[DuckDuckGoTools()], add_history_to_context=True, num_history_runs=3, add_datetime_to_context=True, markdown=True, ) # Create custom FastAPI app app: FastAPI = FastAPI( title="Custom FastAPI App", version="1.0.0", ) # Add your own routes @app.post("/customers") async def get_customers(): return [ { "id": 1, "name": "John Doe", "email": "john.doe@example.com", }, { "id": 2, "name": "Jane Doe", "email": "jane.doe@example.com", }, ] # Set up the AgentOS app by passing your FastAPI app agent_os = AgentOS( description="Example app with custom routers", agents=[web_research_agent], base_app=app, ) # Alternatively, add all routes from AgentOS app to the current app # for route in agent_os.get_routes(): # app.router.routes.append(route) app = agent_os.get_app() if __name__ == "__main__": """Run the AgentOS application. You can see the docs at: http://localhost:7777/docs """ agent_os.serve(app="custom_fastapi_app:app", reload=True) ``` ## Middleware and Dependencies You can add middleware and dependencies to your custom FastAPI app: ```python theme={null} from fastapi import FastAPI, Depends, HTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.security import HTTPBearer app = FastAPI() # Add CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["https://yourdomain.com"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Security dependency security = HTTPBearer() async def verify_token(token: str = Depends(security)): if token.credentials != "your-secret-token": raise HTTPException(status_code=401, detail="Invalid token") return token # Protected route @app.get("/protected", dependencies=[Depends(verify_token)]) async def protected_endpoint(): return {"message": "Access granted"} # Integrate with AgentOS agent_os = AgentOS( agents=[Agent(id="basic-agent", model=OpenAIChat(id="gpt-5-mini"))], base_app=app ) app = agent_os.get_app() ``` ## Access AgentOS Routes You can programmatically access and inspect the routes added by AgentOS: ```python theme={null} agent_os = AgentOS(agents=[agent]) app = agent_os.get_app() # Get all routes routes = agent_os.get_routes() for route in routes: print(f"Route: {route.path}") if hasattr(route, 'methods'): print(f"Methods: {route.methods}") ``` ## Developer Resources * [AgentOS Reference](/reference/agent-os/agent-os) * [Full Example](/agent-os/usage/custom-fastapi) * [FastAPI Documentation](https://fastapi.tiangolo.com/) # Chat Interface Source: https://docs.agno.com/agent-os/features/chat-interface Use AgentOS chat to talk to agents, collaborate with teams, and run workflows ## Overview The AgentOS chat is the home for day‑to‑day work with your AI system. From one screen you can: * Chat with individual agents * Collaborate with agent teams * Trigger and monitor workflows * Review sessions, knowledge, memory, and metrics It’s designed to feel familiar—type a message, attach files, and get live, streaming responses. Each agent, team, and workflow maintains its own context so you can switch between tasks without losing your place. ## Chat Interfaces ### Chat with an Agent * Select an agent from the right panel. * Ask a question like “What tools do you have access to?” * Agents keep their own history, tools, and instructions; switching agents won’t mix contexts. **Learn more about Agents**: Dive deeper into agent configuration, tools, memory, and advanced features in our [Agents Documentation](/basics/agents/overview). ### Work with a Team * Switch the top toggle to Teams and pick a team. * A team delegates tasks to its members and synthesizes their responses into a cohesive response. * Use the chat stream to watch how the team divides and solves the task. **Learn more about Teams**: Explore team modes, coordination strategies, and multi-agent collaboration in our [Teams Documentation](/basics/teams/overview). ### Run a Workflow * Switch to Workflows and choose one. * Provide the input (plain text or structured, depending on the workflow). * Watch execution live: steps stream as they start, produce output, and finish. **Learn more about Workflows**: Discover workflow types, advanced patterns, and automation strategies in our [Workflows Documentation](/basics/workflows/overview). ## Troubleshooting * The page loads but nothing responds: verify your AgentOS app is running. * Can’t see previous chats: you may be in a new session—open the Sessions panel and pick an older one. * File didn’t attach: try a common format (png, jpg, pdf, csv, docx, txt, mp3, mp4) and keep size reasonable. ## Related Examples Comprehensive demo with agents, knowledge, and evaluation system Advanced demo with knowledge, storage, and multiple agents Deploy agents to Slack channels Connect agents to WhatsApp messaging # Knowledge Management Source: https://docs.agno.com/agent-os/features/knowledge-management Upload, organize, and manage knowledge for your agents in AgentOS ## Overview Upload files, add web pages, or paste text to build a searchable knowledge base for your agents. AgentOS indexes content and shows processing status so you can track what’s ready to use. Prerequisites: Your AgentOS must be connected and active. If you see “Disconnected” or “Inactive,” review your connection settings. ## Accessing Knowledge * Open the `Knowledge` section in the sidebar. * If multiple knowledge databases are configured, select one from the database selector in the header. * Use the `Refresh` button to sync status and content. # Knowledge Management * **View Content**: Browse your Knowledge base contents * **Add Content**: Upload new documents, add URLs, or input text directly * **Edit Content**: Modify metadata on existing Knowledge entries * **Delete Content**: Remove outdated content ## What You Can Add * Files: `.pdf`, `.csv`, `.json`, `.txt`, `.doc`, `.docx` , `.md`, `.xlsx`, `.xls`, `.pptx`. * Web: Website URLs (pages) or direct file links * Text: Type or paste content directly Available processing options (Readers and Chunkers) are provided by your OS and may vary by file/URL type. ## Adding Content 1. Start an upload * Click `ADD NEW CONTENT`, then choose `FILE`, `WEB`, or `TEXT`. 2. FILE * Drag & drop or select files. You can also add a file URL. * Add details per item: Name, Description, Metadata, Reader, and optional Chunker. * Names must be unique across items. * Save to upload one or many at once. 3. WEB * Enter one or more URLs and add them to the list. * Add details per item as above (Name, Description, Metadata, Reader/Chunker). * Save to upload all listed URLs. 4. TEXT * Paste or type content. * Set Name, optional Description/Metadata, and Reader/Chunker. * Add Content to upload. ## Useful Links Learn how to add knowledge to agents and use RAG Understand knowledge sharing in team environments # Memories Source: https://docs.agno.com/agent-os/features/memories View and manage persistent memory storage for your agents in AgentOS ## Overview The Memories feature in AgentOS provides a centralized view of information that agents have learned and stored about you as a user. Memory gives agents the ability to recall information about you across conversations, enabling personalized and contextual interactions. * Memories are created and updated during an agent run * Each memory is tied to a specific user ID and contains learned information * Memories include content, topics, timestamps, and the input that generated them * Agents with memory enabled can learn about you and provide more relevant responses over time Prerequisites: Your AgentOS must be connected and active. If you see "Disconnected" or "Inactive," review your connection settings. ## Accessing Memories * Open the `Memory` section in the sidebar. * View all stored memories in a chronological table format. * Click the `Refresh` button to sync the latest memory updates. ## Memory Management The memory interface allows you to: 1. **Create memories** - Create memories your agents can reference during chat sessions 2. **View by topics** - See memories organized by thematic categories 3. **Edit memories** - Update or correct stored information as needed 4. **Delete memories** - Remove outdated or incorrect information 5. **Monitor memory creation** - See when and from what inputs memories were generated Memories are automatically generated from your conversations, but you can also manually create, edit, or remove them. ## Privacy and Control * All memories are tied to a specific user ID and stored in your AgentOS database * Memories are only accessible to agents within your connected OS instance * Memory data remains within your deployment and is never shared externally ## Useful Links Learn how memory works and how agents learn about users Understand data protection and privacy features # Session Tracking Source: https://docs.agno.com/agent-os/features/session-tracking Monitor, analyze, and manage agent sessions through the AgentOS interface ## Overview Sessions are durable conversation timelines that bind inputs, model outputs, tools, files, metrics, and summaries under a single `session_id`. AgentOS persists sessions for Agents, Teams, and Workflows so you can resume work, audit behavior, and analyze quality over time. * A session collects ordered runs (each run contains messages, tool calls, and metrics). * Summaries and metadata help you search, group, and reason about long histories. * Token usage can be monitored per session via the metrics tab. * Inspect details about the agent and models tied to each session. ## Accessing Sessions * Open the `Sessions` section in the sidebar. * If multiple session databases are configured, pick one from the database selector in the header. * Switch between `Agents` and `Teams` using the header tabs. * Click `Reload page` (Refresh) to sync the list and statuses. ## Troubleshooting * Sessions not loading: Ensure your OS is connected and active, select a session database, then click `Reload page`. * No sessions yet: Start a conversation to generate sessions. * Wrong list: Check the `Agents` vs `Teams` tab and sorting. * Configuration errors: If you see endpoint or database errors, verify your OS endpoint and database settings. ## Useful Links Learn about sessions and multi-turn conversations Understand chat history and multi-turn conversations # Tracing Source: https://docs.agno.com/agent-os/features/tracing View and analyze execution traces for your agents and teams in AgentOS ## Overview The Tracing feature in AgentOS provides deep observability into your agent and team executions. Every run generates a trace containing detailed information about what happened, how long it took, and what resources were used. * Traces capture the complete execution flow of agents and teams * Each trace contains spans representing individual operations (LLM calls, tool executions, etc.) * Each span contains token usage (when relevant), latency, and error information at a glance * You can filter and search traces to debug issues or analyze performance Prerequisites: Your AgentOS must be connected and active with tracing enabled. If you see "Disconnected" or "Inactive," review your connection settings. ## Accessing Tracing * Open the `Tracing` section in the sidebar. * Click the `Refresh` button to sync the latest traces. ## Filtering Traces Use the filtering options to quickly find the traces you need: ### View Mode * **By Session** - Group traces by conversation session to see all activity within a session * **By Run** - View individual execution runs to analyze specific interactions ### Time Range * **Quick filters** - Select from preset ranges: last hour, last 24 hours, or last 7 days * **Custom range** - Pick specific start and end dates for precise time-based filtering Traces date filter options ## Trace Views Click on any trace to view its execution breakdown. You can switch between two visualization modes: ### Tree View The Tree view displays spans in a hierarchical structure, showing the parent-child relationships between operations. This is useful for understanding how teams delegate to agents and how agents invoke tools. ### Waterfall View The Waterfall view visualizes spans on a time axis, showing when each operation started, how long it took, and which operations ran in parallel. This helps identify bottlenecks and optimize performance. ## Useful Links Learn about tracing concepts and how traces are structured Configure tracing for your AgentOS deployment Explore the tracing API endpoints # A2A Source: https://docs.agno.com/agent-os/interfaces/a2a/introduction Expose Agno agents via the A2A protocol v2.1.2 Google's [Agent-to-Agent Protocol (A2A)](https://a2a-protocol.org/latest/topics/what-is-a2a/) provides a standard way for agents to communicate with each other. Agno integrates with A2A, enabling Agno agents and teams to be exposed in an A2A-compatible format. The `A2A` interface works with the [AgentOS](/agent-os/introduction) runtime to provide this functionality. ## Setup Set `a2a_interface=True` when creating an `AgentOS` instance: ```python a2a_agentos.py theme={null} from agno.agent import Agent from agno.os import AgentOS from agno.os.interfaces.a2a import A2A agent = Agent(name="My Agno Agent",id="my_agent") agent_os = AgentOS( agents=[agent], a2a_interface=True, ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="a2a:app", reload=True) ``` By default, all agents, teams, and workflows in the AgentOS are exposed via A2A. Specific agents, teams, and workflows can be exposed by initializing the interface explicitly: ```python a2a-interface-initialization.py theme={null} from agno.agent import Agent from agno.os import AgentOS from agno.os.interfaces.a2a import A2A agent = Agent(name="My Agno Agent",id="my_agent") # Initialize the A2A interface specifying the agents to expose a2a = A2A(agents=[agent]) agent_os = AgentOS( agents=[agent], interfaces=[a2a], # Pass the A2A interface to the AgentOS using the `interfaces` parameter ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="a2a-interface-initialization:app", reload=True) ``` ## A2A Endpoints For each available agent, team and workflow, the following A2A-compatible endpoints will be available: ### Agents * `/a2a/agents/{id}/.well-known/agent-card.json`: Returns the [Agent Card](https://a2a-protocol.org/v0.3.0/topics/agent-discovery/#1-well-known-uri) describing the agent in A2A format. See [API reference](/reference-api/schema/a2a/get-agent-card). * `/a2a/agents/{id}/v1/message:stream`: Runs the agent, streaming the responses as events in A2A format. See [API reference](/reference-api/schema/a2a/stream-message-agent) and [A2A protocol docs](https://a2a-protocol.org/v0.3.0/specification/#356-method-mapping-reference-table). * `/a2a/agents/{id}/v1/message:send`: Runs the agen, returning the response in A2A format (non-streaming). See [A2A protocol docs](https://a2a-protocol.org/v0.3.0/specification/#356-method-mapping-reference-table). ### Teams * `/a2a/teams/{id}/.well-known/agent-card.json`: Returns the Team Card describing the Team in A2A format. See [API reference](/reference-api/schema/a2a/get-agent-card). * `/a2a/teams/{id}/v1/message:stream`: Runs the team, streaming the responses as events in A2A format. See [API reference](/reference-api/schema/a2a/stream-message-agent) and [A2A protocol docs](https://a2a-protocol.org/v0.3.0/specification/#356-method-mapping-reference-table). * `/a2a/teams/{id}/v1/message:send`: Runs the team, returning the response in A2A format (non-streaming). See [API reference](/reference-api/schema/a2a/run-message-agent). ### Workflows * `/a2a/workflows/{id}/.well-known/agent-card.json`: Returns the Workflow Card describing the Workflow in A2A format. See [API reference](/reference-api/schema/a2a/get-agent-card). * `/a2a/workflows/{id}/v1/message:stream`: Runs the workflow, streaming the responses as events in A2A format. See [API reference](/reference-api/schema/a2a/stream-message-agent) and [A2A protocol docs](https://a2a-protocol.org/v0.3.0/specification/#356-method-mapping-reference-table). * `/a2a/workflows/{id}/v1/message:send`: Runs the workflow, returning the response in A2A format (non-streaming). See [API reference](/reference-api/schema/a2a/run-message-agent). A2A clients expect a server to expose only a single agent. To use your Agno A2A interface with those clients, simply use `/a2a/agents/{id}/` (or `/a2a/teams/{id}/`, `/a2a/workflows/{id}/`) as the base url. ## Developer Resources * View [AgentOS Reference](/reference/agent-os/agent-os) * View [A2A Documentation](https://a2a-protocol.org/latest/) * View [Examples](/agent-os/usage/interfaces/a2a/basic) * View [Cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/06_agent_os/interfaces/a2a) # AG-UI Source: https://docs.agno.com/agent-os/interfaces/ag-ui/introduction Expose Agno agents via the AG-UI protocol AG-UI, the [Agent-User Interaction Protocol](https://github.com/ag-ui-protocol/ag-ui), standardizes how AI agents connect to frontend applications. **Migration from Apps**: For migration from `AGUIApp`, see the [v2 migration guide](/how-to/v2-migration#7-apps-interfaces) for complete steps. ## Example usage ```bash theme={null} pip install ag-ui-protocol ``` Expose an Agno agent through the AG-UI interface using `AgentOS` and `AGUI`. ```python basic.py theme={null} from agno.agent.agent import Agent from agno.models.openai import OpenAIChat from agno.os import AgentOS from agno.os.interfaces.agui import AGUI chat_agent = Agent(model=OpenAIChat(id="gpt-4o")) agent_os = AgentOS(agents=[chat_agent], interfaces=[AGUI(agent=chat_agent)]) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="basic:app", reload=True) ``` Use Dojo (`ag-ui`'s frontend) as an advanced, customizable interface for AG-UI agents. 1. Clone: `git clone https://github.com/ag-ui-protocol/ag-ui.git` 2. Install dependencies in `/ag-ui/typescript-sdk`: `pnpm install` 3. Build the Agno package in `/ag-ui/integrations/agno`: `pnpm run build` 4. Start Dojo following the instructions in the repository. With Dojo running, open `http://localhost:3000` and select the Agno agent. Additional examples are available in the [cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/06_agent_os/interfaces/agui/). ## Custom Events Custom events created in tools are automatically delivered to AG-UI in the AG-UI custom event format. **Creating custom events:** ```python theme={null} from dataclasses import dataclass from agno.run.agent import CustomEvent @dataclass class CustomerProfileEvent(CustomEvent): customer_name: str customer_email: str ``` **Yielding from tools:** ```python theme={null} from agno.tools import tool @tool() async def get_customer_profile(customer_id: str): customer = fetch_customer(customer_id) yield CustomerProfileEvent( customer_name=customer["name"], customer_email=customer["email"], ) return f"Profile retrieved for {customer['name']}" ``` Custom events are streamed in real-time to the AG-UI frontend. See [Custom Events documentation](/basics/agents/running-agents#custom-events) for more details. ## Core Components * `AGUI` (interface): Wraps an Agno `Agent` or `Team` into an AG-UI compatible FastAPI router. * `AgentOS.serve`: Serves the FastAPI app (including the AGUI router) with Uvicorn. `AGUI` mounts protocol-compliant routes on the app. ## `AGUI` interface Main entry point for AG-UI exposure. ### Initialization Parameters | Parameter | Type | Default | Description | | --------- | ----------------- | ------- | ---------------------- | | `agent` | `Optional[Agent]` | `None` | Agno `Agent` instance. | | `team` | `Optional[Team]` | `None` | Agno `Team` instance. | Provide `agent` or `team`. ### Key Method | Method | Parameters | Return Type | Description | | ------------ | ------------------------ | ----------- | -------------------------------------------------------- | | `get_router` | `use_async: bool = True` | `APIRouter` | Returns the AG-UI FastAPI router and attaches endpoints. | ## Endpoints Mounted at the interface's route prefix (root by default): * `POST /agui`: Main entrypoint. Accepts `RunAgentInput` from `ag-ui-protocol`. Streams AG-UI events. * `GET /status`: Health/status endpoint for the interface. Refer to `ag-ui-protocol` docs for payload details. ## Serving AgentOS Use `AgentOS.serve` to run the app with Uvicorn. ### Parameters | Parameter | Type | Default | Description | | --------- | --------------------- | ------------- | -------------------------------------- | | `app` | `Union[str, FastAPI]` | required | FastAPI app instance or import string. | | `host` | `str` | `"localhost"` | Host to bind. | | `port` | `int` | `7777` | Port to bind. | | `reload` | `bool` | `False` | Enable auto-reload for development. | See [cookbook examples](https://github.com/agno-agi/agno/tree/main/cookbook/06_agent_os/interfaces/agui/) for updated interface patterns. # Interfaces Source: https://docs.agno.com/agent-os/interfaces/overview Expose Agno agents through various communication protocols and platforms Interfaces enable exposing Agno agents, teams, and workflows through various communication protocols and platforms. Each interface provides a standardized way to connect Agno agents to external systems, messaging platforms, and frontend applications. ## Available Interfaces Connect agents to frontend applications using the Agent-User Interaction Protocol Deploy agents as Slack applications for team collaboration Serve agents via WhatsApp for direct messaging interactions Expose agents via the Agent-to-Agent Protocol for inter-agent communication ## How Interfaces Work Interfaces are FastAPI routers that mount protocol-specific endpoints on an AgentOS instance. Each interface: * Wraps Agno agents, teams, or workflows into protocol-compatible endpoints * Handles authentication and request validation for the target platform * Manages session tracking and context preservation * Streams responses back to clients in the appropriate format ## Using Interfaces Interfaces are added to an AgentOS instance through the `interfaces` parameter: ```python theme={null} from agno.os import AgentOS from agno.os.interfaces.slack import Slack agent_os = AgentOS( agents=[my_agent], interfaces=[Slack(agent=my_agent)], ) app = agent_os.get_app() ``` Multiple interfaces can be added to a single AgentOS instance, allowing the same agents to be exposed through different protocols simultaneously. # Slack Source: https://docs.agno.com/agent-os/interfaces/slack/introduction Host agents as Slack Applications. Use the Slack interface to serve Agents, Teams, or Workflows on Slack. It mounts Slack event routes on a FastAPI app and sends responses back to Slack threads. ## Setup Steps Follow the Slack setup guide in the [cookbook](https://github.com/agno-agi/agno/blob/main/cookbook/06_agent_os/interfaces/slack/README.md). Required configuration: * `SLACK_TOKEN` (Bot User OAuth Token) * `SLACK_SIGNING_SECRET` (App Signing Secret) * An ngrok tunnel (for local development) and event subscriptions pointing to `/slack/events` ## Example Usage Create an agent, expose it with the `Slack` interface, and serve via `AgentOS`: ```python basic.py theme={null} from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.os import AgentOS from agno.os.interfaces.slack import Slack basic_agent = Agent( name="Basic Agent", model=OpenAIChat(id="gpt-5-mini"), # Ensure OPENAI_API_KEY is set add_history_to_context=True, num_history_runs=3, add_datetime_to_context=True, ) agent_os = AgentOS( agents=[basic_agent], interfaces=[Slack(agent=basic_agent)], ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="basic:app", port=8000, reload=True) ``` A complete example is available at `cookbook/06_agent_os/interfaces/slack/basic.py`. ## Core Components * `Slack` (interface): Wraps an Agno `Agent`, `Team`, or `Workflow` for Slack integration via FastAPI. * `AgentOS.serve`: Serves the FastAPI app using Uvicorn. ## `Slack` Interface Main entry point for Agno Slack applications. ### Initialization Parameters | Parameter | Type | Default | Description | | ------------------------ | --------------------- | ---------- | ----------------------------------------------------------------------------------------------------------------------------------------- | | `agent` | `Optional[Agent]` | `None` | Agno `Agent` instance. | | `team` | `Optional[Team]` | `None` | Agno `Team` instance. | | `workflow` | `Optional[Workflow]` | `None` | Agno `Workflow` instance. | | `prefix` | `str` | `"/slack"` | Custom FastAPI route prefix for the Slack interface. | | `tags` | `Optional[List[str]]` | `None` | FastAPI route tags for API documentation. Defaults to `["Slack"]` if not provided. | | `reply_to_mentions_only` | `bool` | `True` | When `True` (default), bot responds to @mentions in channels and all direct messages. When `False`, responds to all messages in channels. | Provide `agent`, `team`, or `workflow`. ### Key Method | Method | Parameters | Return Type | Description | | ------------ | ---------- | ----------- | -------------------------------------------------- | | `get_router` | None | `APIRouter` | Returns the FastAPI router and attaches endpoints. | ## Endpoints Mounted under the `/slack` prefix: ### `POST /slack/events` * Handles all Slack events (URL verification, messages, app mentions) * Verifies Slack signature on each request * Uses thread timestamps as session IDs for per-thread context * Streams responses back into the originating thread (splits long messages) ## Testing the Integration 1. Run the app locally: `python .py` (ensure ngrok is running) 2. Invite the bot to a channel: `/invite @YourAppName` 3. Mention the bot in a channel: `@YourAppName hello` 4. Open a DM with the bot and send a message ## Troubleshooting * Verify `SLACK_TOKEN` and `SLACK_SIGNING_SECRET` are set * Confirm the bot is installed and invited to the channel * Check ngrok URL and event subscription path (`/slack/events`) * Review application logs for signature failures or permission errors # WhatsApp Source: https://docs.agno.com/agent-os/interfaces/whatsapp/introduction Host agents as WhatsApp applications Use the WhatsApp interface to serve Agents or Teams via WhatsApp. It mounts webhook routes on a FastAPI app and sends responses back to WhatsApp users and threads. ## Setup Follow the WhatsApp setup guide in the [Whatsapp Cookbook](https://github.com/agno-agi/agno/blob/main/cookbook/06_agent_os/interfaces/whatsapp/readme.md). Required environment variables: * `WHATSAPP_ACCESS_TOKEN` * `WHATSAPP_PHONE_NUMBER_ID` * `WHATSAPP_VERIFY_TOKEN` * Optional (production): `WHATSAPP_APP_SECRET` and `APP_ENV=production` The user's phone number is automatically used as the `user_id` for runs. This ensures that sessions and memory are appropriately scoped to the user. The phone number is also used for the `session_id`, so a single WhatsApp conversation corresponds to a single session. This should be considered when managing session history. ## Example Usage Create an agent, expose it with the `Whatsapp` interface, and serve via `AgentOS`: ```python theme={null} from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.os import AgentOS from agno.os.interfaces.whatsapp import Whatsapp image_agent = Agent( model=OpenAIChat(id="gpt-5-mini"), # Ensure OPENAI_API_KEY is set tools=[OpenAITools(image_model="gpt-image-1")], markdown=True, add_history_to_context=True, ) agent_os = AgentOS( agents=[image_agent], interfaces=[Whatsapp(agent=image_agent)], ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="basic:app", port=8000, reload=True) ``` Additional examples are available in the [cookbook](https://github.com/agno-agi/agno/tree/main/cookbook/06_agent_os/interfaces/whatsapp/). ## Core Components * `Whatsapp` (interface): Wraps an Agno `Agent` or `Team` for WhatsApp via FastAPI. * `AgentOS.serve`: Serves the FastAPI app using Uvicorn. ## `Whatsapp` Interface Main entry point for Agno WhatsApp applications. ### Initialization Parameters | Parameter | Type | Default | Description | | --------- | ----------------- | ------- | ---------------------- | | `agent` | `Optional[Agent]` | `None` | Agno `Agent` instance. | | `team` | `Optional[Team]` | `None` | Agno `Team` instance. | Provide `agent` or `team`. ### Key Method | Method | Parameters | Return Type | Description | | ------------ | ------------------------ | ----------- | -------------------------------------------------- | | `get_router` | `use_async: bool = True` | `APIRouter` | Returns the FastAPI router and attaches endpoints. | ## Endpoints Mounted under the `/whatsapp` prefix: ### `GET /whatsapp/status` * Health/status of the interface. ### `GET /whatsapp/webhook` * Verifies WhatsApp webhook (`hub.challenge`). * Returns `hub.challenge` on success; `403` on token mismatch; `500` if `WHATSAPP_VERIFY_TOKEN` missing. ### `POST /whatsapp/webhook` * Receives WhatsApp messages and events. * Validates signature (`X-Hub-Signature-256`); bypassed in development mode. * Processes text, image, video, audio, and document messages via the agent/team. * Sends replies (splits long messages; uploads and sends generated images). * Responses: `200 {"status": "processing"}` or `{"status": "ignored"}`, `403` invalid signature, `500` errors. # What is AgentOS? Source: https://docs.agno.com/agent-os/introduction The production runtime and control plane for your agentic systems AgentOS is Agno's production-ready runtime that runs entirely within your own infrastructure, ensuring complete data privacy and control of your agentic system. Agno also provides a beautiful web interface for managing, monitoring, and interacting with your AgentOS, with no data ever being persisted outside of your environment. Behind the scenes, AgentOS is a FastAPI app that you can run locally or in your cloud. It is designed to be easy to deploy and scale. ## Getting Started Ready to get started with AgentOS? Here's what you need to do: Set up a new AgentOS instance from scratch using our templates Learn how to connect your local development environment to the platform ## Security & Privacy First AgentOS is designed with enterprise security and data privacy as foundational principles, not afterthoughts. AgentOS Security and Privacy Architecture ### Complete Data Ownership * **Your Infrastructure, Your Data**: AgentOS runs entirely within your cloud environment * **Zero Data Transmission**: No conversations, logs, or metrics are sent to external services * **Private by Default**: All processing, storage, and analytics happen locally To learn more about AgentOS Security, check out the [AgentOS Security](/agent-os/security) page. ## Learn more Learn how to use the AgentOS control plane to manage and monitor your OSs Get started by creating your first AgentOS instance # Filter Knowledge Source: https://docs.agno.com/agent-os/knowledge/filter-knowledge Learn how to use advanced filter expressions through the Agno API for precise knowledge base filtering. When using the AgentOS API, you can apply filters to precisely control which knowledge base documents your agents search, without changing your agent code. Filter expressions serialize to JSON and are automatically reconstructed server-side for powerful, programmatic filtering. ## Two Approaches to Filtering Agno supports two ways to filter knowledge through the API: * Use **dictionary filters** for simple "field = value" lookups * Use **filter expressions** when you need OR/NOT logic or ranges ### 1. Dictionary Filters (Simple) Best for straightforward equality matching. Send a JSON object with key-value pairs: ```json theme={null} {"docs": "agno", "status": "published"} ``` ### 2. Filter Expressions (Advanced) Best for complex filtering with full logical control. Send structured filter objects: ```json theme={null} {"op": "AND", "conditions": [ {"op": "EQ", "key": "docs", "value": "agno"}, {"op": "GT", "key": "version", "value": 2} ]} ``` **When to use which:** * Use **dict filters** for simple queries like filtering by category or status * Use **filter expressions** when you need OR logic, exclusions (NOT), or range queries (GT/LT) ## Filter Operators Filter expressions support a range of comparison and logical operators: ### Comparison Operators * **`EQ(key, value)`** - Equality: field equals value * **`GT(key, value)`** - Greater than: field > value * **`LT(key, value)`** - Less than: field \< value * **`IN(key, [values])`** - Inclusion: field in list of values ### Logical Operators * **`AND(*filters)`** - All conditions must be true * **`OR(*filters)`** - At least one condition must be true * **`NOT(filter)`** - Negate a condition ## Serialization Format Filter expression objects use a dictionary format with an `"op"` key that distinguishes them from regular dict filters: ```python theme={null} from agno.filters import EQ, GT, AND # Python filter expression filter_expr = AND(EQ("status", "published"), GT("views", 1000)) # Serialized to JSON { "op": "AND", "conditions": [ {"op": "EQ", "key": "status", "value": "published"}, {"op": "GT", "key": "views", "value": 1000} ] } ``` The presence of the `"op"` key tells the API to deserialize the filter as a filter expression. Regular dict filters (without `"op"`) continue to work for backward compatibility. ## Using Filters Through the API In all examples below, you pass the serialized JSON string via the `knowledge_filters` field when creating a run. ### Dictionary Filters (Simple Approach) For basic filtering, send a JSON object with key-value pairs. All conditions are combined with AND logic: ```python Python Client theme={null} import requests import json # Simple dict filter filter_dict = {"docs": "agno", "status": "published"} # Serialize to JSON filter_json = json.dumps(filter_dict) # Send request response = requests.post( "http://localhost:7777/agents/agno-knowledge-agent/runs", data={ "message": "What are agno's key features?", "stream": "false", "knowledge_filters": filter_json, } ) result = response.json() ``` ```bash cURL theme={null} curl -X 'POST' \ 'http://localhost:7777/agents/agno-knowledge-agent/runs' \ -H 'accept: application/json' \ -H 'Content-Type: multipart/form-data' \ -F 'message=What are agno'\''s key features?' \ -F 'stream=false' \ -F 'session_id=' \ -F 'user_id=' \ -F 'knowledge_filters={"docs": "agno"}' ``` **More Dict Filter Examples:** ```python theme={null} # Filter by single field {"category": "technology"} # Filter by multiple fields (AND logic) {"category": "technology", "status": "published", "year": 2024} # Filter with different data types {"active": True, "priority": 1, "department": "engineering"} ``` ### Filter Expressions (Advanced Approach) For complex filtering with logical operators and comparisons: ```python Python Client theme={null} import requests import json from agno.filters import EQ # Create filter expression filter_expr = EQ("category", "technology") # Serialize to JSON filter_json = json.dumps(filter_expr.to_dict()) # Send request response = requests.post( "http://localhost:7777/agents/my-agent/runs", data={ "message": "What are the latest tech articles?", "stream": "false", "knowledge_filters": filter_json, } ) result = response.json() ``` ```bash cURL theme={null} curl -X 'POST' \ 'http://localhost:7777/agents/my-agent/runs' \ -H 'Content-Type: multipart/form-data' \ -F 'message=What are the latest tech articles?' \ -F 'stream=false' \ -F 'knowledge_filters={"op": "EQ", "key": "category", "value": "technology"}' ``` ### Multiple Filter Expressions Send multiple filter expressions as a JSON array: ```python Python Client theme={null} from agno.filters import EQ, GT # Create multiple filters filters = [ EQ("status", "published"), GT("date", "2024-01-01") ] # Serialize list to JSON filters_json = json.dumps([f.to_dict() for f in filters]) response = requests.post( "http://localhost:7777/agents/my-agent/runs", data={ "message": "Show recent published articles", "stream": "false", "knowledge_filters": filters_json, } ) ``` ```bash cURL theme={null} curl -X 'POST' \ 'http://localhost:7777/agents/my-agent/runs' \ -H 'Content-Type: multipart/form-data' \ -F 'message=Show recent published articles' \ -F 'stream=false' \ -F 'knowledge_filters=[{"op": "EQ", "key": "status", "value": "published"}, {"op": "GT", "key": "date", "value": "2024-01-01"}]' ``` ## Error Handling ### Invalid Filter Structure When filters have errors, they're gracefully ignored with warnings: ```bash theme={null} # Missing required fields curl ... -F 'knowledge_filters={"op": "EQ", "key": "status"}' # Result: Filter ignored, warning logged # Unknown operator curl ... -F 'knowledge_filters={"op": "UNKNOWN", "key": "status", "value": "x"}' # Result: Filter ignored, warning logged # Invalid JSON curl ... -F 'knowledge_filters={invalid json}' # Result: Filter ignored, warning logged ``` When filters fail to parse, the search proceeds **without filters** rather than throwing an error. Always verify your filter JSON is valid and check server logs if results seem unfiltered. ### Client-Side Validation Add validation before sending requests: ```python theme={null} def validate_and_send_filter(filter_expr, message): """Validate filter before sending to API.""" try: # Test serialization filter_dict = filter_expr.to_dict() filter_json = json.dumps(filter_dict) # Verify it's valid JSON json.loads(filter_json) # Send request return send_filtered_agent_request(message, filter_expr) except (AttributeError, TypeError, json.JSONDecodeError) as e: print(f"Filter validation failed: {e}") return None ``` ## Next Steps Learn about filter expressions and metadata design in detail Explore the full Agent OS API reference Understand knowledge base architecture and setup Optimize your search strategies See working examples of filter expressions via API # Manage Knowledge Source: https://docs.agno.com/agent-os/knowledge/manage-knowledge Attach Knowledge to your AgentOS instance The AgentOS control plane provides a simple way to [manage](/agent-os/features/knowledge-management#adding-content) your Knowledge bases. You can add, edit, and delete content from your Knowledge bases directly through the control plane. You can specify multiple Knowledge bases and reuse the same Knowledge instance across different Agents or Teams as needed. ## Prerequisites Before setting up Knowledge management in AgentOS, ensure you have: * PostgreSQL database running and accessible - used for this example * Required dependencies installed: `pip install agno` * OpenAI API key configured (for embeddings) * Basic understanding of [Knowledge concepts](/basics/knowledge/getting-started) ## Example This example demonstrates how to attach multiple Knowledge bases to AgentOS and populate them with content from different sources. ```python agentos_knowledge.py theme={null} from textwrap import dedent from agno.db.postgres import PostgresDb from agno.knowledge.embedder.openai import OpenAIEmbedder from agno.knowledge.knowledge import Knowledge from agno.os import AgentOS from agno.vectordb.pgvector import PgVector, SearchType # ************* Setup Knowledge Databases ************* db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" documents_db = PostgresDb( db_url, id="agno_knowledge_db", knowledge_table="agno_knowledge_contents", ) faq_db = PostgresDb( db_url, id="agno_faq_db", knowledge_table="agno_faq_contents", ) # ******************************* documents_knowledge = Knowledge( vector_db=PgVector( db_url=db_url, table_name="agno_knowledge_vectors", search_type=SearchType.hybrid, embedder=OpenAIEmbedder(id="text-embedding-3-small"), ), contents_db=documents_db, ) faq_knowledge = Knowledge( vector_db=PgVector( db_url=db_url, table_name="agno_faq_vectors", search_type=SearchType.hybrid, embedder=OpenAIEmbedder(id="text-embedding-3-small"), ), contents_db=faq_db, ) agent_os = AgentOS( description="Example app with AgentOS Knowledge", # Add the knowledge bases to AgentOS knowledge=[documents_knowledge, faq_knowledge], ) app = agent_os.get_app() if __name__ == "__main__": documents_knowledge.add_content( name="Agno Docs", url="https://docs.agno.com/llms-full.txt", skip_if_exists=True ) faq_knowledge.add_content( name="Agno FAQ", text_content=dedent(""" What is Agno? Agno is a framework for building agents. Use it to build multi-agent systems with memory, knowledge, human in the loop and MCP support. """), skip_if_exists=True, ) # Run your AgentOS # You can test your AgentOS at: http://localhost:7777/ agent_os.serve(app="agentos_knowledge:app") ``` ### Screenshots The screenshots below show how you can access and manage your different Knowledge bases through the AgentOS interface: llm-app-aidev-run llm-app-aidev-run ## Best Practices * **Separate Knowledge by Domain**: Create separate Knowledge bases for different topics (e.g., technical docs, FAQs, policies) * **Consistent Naming**: Use descriptive names for your Knowledge bases that reflect their content * **Regular Updates**: Keep your Knowledge bases current by regularly adding new content and removing outdated information * **Monitor Performance**: Use different table names for vector storage to avoid conflicts * **Content Organization**: Use the `name` parameter when adding content to make it easily identifiable * **Use metadata for filtering and searching**: Add metadata to your content to make it easier to find and filter ## Troubleshooting Ensure your knowledge base is properly added to the `knowledge` parameter when creating your AgentOS instance. Also make sure to attach a `contents_db` to your Knowledge instance. Verify your PostgreSQL connection string and ensure the database is running and accessible. Check that your content has been properly embedded by verifying entries in your vector database table. # Custom Lifespan Source: https://docs.agno.com/agent-os/lifespan Customize the lifespan of your AgentOS app to handle startup and shutdown logic. You will often want to run code before your AgentOS app starts or before it shuts down. This can be done by providing a custom **lifespan function** via the `lifespan` parameter. This is how a lifespan function looks like: ```python theme={null} @asynccontextmanager async def lifespan(app): # This will run before your app starts log_info("Starting My FastAPI App") yield # This will run before your app shuts down log_info("Stopping My FastAPI App") ``` ## FastAPI Lifespan The custom lifespan function you provide will be used as the **lifespan context manager** for the FastAPI app used by your AgentOS. Remember to decorate it with `@asynccontextmanager` as shown in the examples. See the [FastAPI documentation](https://fastapi.tiangolo.com/advanced/events/#lifespan-events) for more information about the lifespan context manager. If you are using a custom FastAPI app, you don't need to worry about overwriting its lifespan. The lifespan you provide will wrap the existing lifespan of the app, letting you combine all. ## Common Use Cases Lifespan control is useful to handle typical startup and shutdown tasks, such as: * **Resource Initialization**: databases, third party services, caches... or anything else needed by your app. * **Cleanup**: Close connections, store data or release resources before shut down. * **Health Checks**: Verify dependencies are available before serving requests * **Background Tasks**: Start/stop background processes ## Example ```python custom_lifespan.py theme={null} from contextlib import asynccontextmanager from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.anthropic import Claude from agno.os import AgentOS from agno.utils.log import log_info # Setup the database db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") # Setup basic agents, teams and workflows agno_support_agent = Agent( id="example-agent", name="Example Agent", model=Claude(id="claude-sonnet-4-0"), db=db, markdown=True, ) @asynccontextmanager async def lifespan(app): log_info("Starting My FastAPI App") yield log_info("Stopping My FastAPI App") agent_os = AgentOS( description="Example app with custom lifespan", agents=[agno_support_agent], lifespan=lifespan, ) app = agent_os.get_app() if __name__ == "__main__": """Run your AgentOS. You can see test your AgentOS at: http://localhost:7777/docs """ agent_os.serve(app="custom_lifespan:app") ``` ```bash theme={null} export ANTHROPIC_API_KEY=your_anthropic_api_key ``` ```bash theme={null} pip install -U agno anthropic "fastapi[standard]" uvicorn sqlalchemy ``` ```bash theme={null} # Using Docker docker run -d \ --name agno-postgres \ -e POSTGRES_DB=ai \ -e POSTGRES_USER=ai \ -e POSTGRES_PASSWORD=ai \ -p 5532:5432 \ pgvector/pgvector:pg17 ``` ```bash FastAPI CLI theme={null} fastapi run custom_lifespan.py ``` ```bash Mac theme={null} python custom_lifespan.py ``` ```bash Windows theme={null} python ovecustom_lifespanrride_routes.py ``` # AgentOS as MCP Server Source: https://docs.agno.com/agent-os/mcp/mcp Learn how and why to expose your AgentOS as an MCP server Your AgentOS will by default be exposed as an API. But you can also expose it as an **MCP server**. This is done by setting `enable_mcp_server=True` when creating your AgentOS instance: ```python theme={null} agent_os = AgentOS(enable_mcp_server=True) ``` ## Why use MCP? The [MCP protocol](https://modelcontextprotocol.io) has become the industry standard to handle connecting AI applications with external tools and data sources. By exposing your AgentOS as an MCP server, external clients that can handle MCP-compatible applications will be able to connect to your AgentOS and interact with it. ## Example ```python enable_mcp_example.py theme={null} from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.anthropic import Claude from agno.os import AgentOS from agno.tools.duckduckgo import DuckDuckGoTools # Setup the database db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") # Create your agents web_research_agent = Agent( name="Web Research Agent", model=Claude(id="claude-sonnet-4-0"), db=db, tools=[DuckDuckGoTools()], markdown=True, ) # Setup AgentOS with MCP enabled agent_os = AgentOS( description="Example app with MCP enabled", agents=[web_research_agent], enable_mcp_server=True, # This enables a LLM-friendly MCP server at /mcp ) app = agent_os.get_app() if __name__ == "__main__": # Your MCP server will be available at http://localhost:7777/mcp agent_os.serve(app="enable_mcp_example:app", reload=True) ``` Once enabled, your AgentOS will expose an MCP server via the `/mcp` endpoint. You can see a complete example [here](/agent-os/usage/mcp/enable-mcp-example). ## Available MCP Tools When you expose your AgentOS as an MCP server, the following MCP tools will be available: ### `get_agentos_config` Get the configuration of the AgentOS ### `run_agent` Run the desired Agent * `agent_id` (str): The ID of the agent to run * `message` (str): The message to send to the agent ### `run_team` Run the desired Team * `team_id` (str): The ID of the team to run * `message` (str): The message to send to the team ### `run_workflow` Run the desired Workflow * `workflow_id` (str): The ID of the workflow to run * `message` (str): The message to send to the workflow ### `get_sessions_for_agent` Get the list of sessions for the desired Agent * `agent_id` (str): The ID of the agent to get the sessions for * `db_id` (str): The ID of the database to use * `user_id` (Optional\[str]): The ID of the user to get the sessions for * `sort_by` (Optional\[str]): The field to sort the sessions by. Defaults to `created_at` * `sort_order` (Optional\[str]): The order to sort the sessions by. Defaults to `desc` ### `get_sessions_for_team` Get the list of sessions for the desired Team * `team_id` (str): The ID of the team to get the sessions for * `db_id` (str): The ID of the database to use * `user_id` (Optional\[str]): The ID of the user to get the sessions for * `sort_by` (Optional\[str]): The field to sort the sessions by. Defaults to `created_at` * `sort_order` (Optional\[str]): The order to sort the sessions by. Defaults to `desc` ### `get_sessions_for_workflow` Get the list of sessions for the desired Workflow * `workflow_id` (str): The ID of the workflow to get the sessions for * `db_id` (str): The ID of the database to use * `user_id` (Optional\[str]): The ID of the user to get the sessions for * `sort_by` (Optional\[str]): The field to sort the sessions by. Defaults to `created_at` * `sort_order` (Optional\[str]): The order to sort the sessions by. Defaults to `desc` ### `create_memory` Create a new user memory * `db_id` (str): The ID of the database to use * `memory` (str): The memory content to store * `user_id` (str): The user this memory is about * `topics` (Optional\[list\[str]]): The topics of the memory ### `get_memories_for_user` Get the list of memories for the given user * `user_id` (str): The ID of the user to get the memories for * `db_id` (Optional\[str]): The ID of the database to use * `sort_by` (Optional\[str]): The field to sort the memories by. Defaults to `created_at` * `sort_order` (Optional\[str]): The order to sort the memories by. Defaults to `desc` ### `update_memory` Update the desired memory * `db_id` (str): The ID of the database to use * `memory_id` (str): The ID of the memory to update * `memory` (str): The memory content to store * `user_id` (str): The ID of the user to update the memory for ### `delete_memory` Delete the desired memory * `db_id` (str): The ID of the database to use * `memory_id` (str): The ID of the memory to delete See a full example [here](/agent-os/usage/mcp/enable-mcp-example). # MCPTools within AgentOS Source: https://docs.agno.com/agent-os/mcp/tools Learn how to use MCPTools in the Agents, Teams and Workflows within your AgentOS The [Model Context Protocol (MCP)](https://modelcontextprotocol.io) enables Agents to interact with external systems through a standardized interface. You can give your Agents access to MCP tools using the `MCPTools` class. Read more about using MCP tools [here](/basics/tools/mcp/overview). Your `MCPTools` will work normally within AgentOS. Their lifecycle is automatically handled, you don't need to handle their connection and disconnection. If you are using `MCPTools` within AgentOS, you should not use `reload=True` when serving your AgentOS. This can break the MCP connection during the FastAPI lifecycle. ## Example ```python mcp_tools_example.py theme={null} from agno.agent import Agent from agno.os import AgentOS from agno.tools.mcp import MCPTools # Create MCPTools instance mcp_tools = MCPTools( transport="streamable-http", url="https://docs.agno.com/mcp" ) # Create MCP-enabled agent agent = Agent( id="agno-agent", name="Agno Agent", tools=[mcp_tools], ) # AgentOS manages MCP lifespan agent_os = AgentOS( description="AgentOS with MCP Tools", agents=[agent], ) app = agent_os.get_app() if __name__ == "__main__": # Don't use reload=True with MCP tools to avoid lifespan issues agent_os.serve(app="mcp_tools_example:app") ``` Refreshing the connection to MCP servers is **not** automatically handled. You can use [`refresh_connection`](/basics/tools/mcp/overview#connection-refresh) to manually refresh a connection. See here for a [full example](/agent-os/usage/mcp/mcp-tools-example). # Custom Middleware Source: https://docs.agno.com/agent-os/middleware/custom Create custom middleware for rate limiting, logging, security, and monitoring in AgentOS v2.1.0 Each middleware wraps your application to intercept requests and responses, enabling you to implement cross-cutting concerns like authentication, logging, and rate limiting. AgentOS supports any [FastAPI/Starlette middleware](https://fastapi.tiangolo.com/tutorial/middleware/). You can create custom middleware for logging, rate limiting, monitoring, security, and more. ## Creating Custom Middleware Middleware in AgentOS follows the FastAPI/Starlette pattern using `BaseHTTPMiddleware`. See the following common middleware examples: ```python Rate Limiting theme={null} """ Rate limiting middleware that limits requests per IP address """ import time from collections import defaultdict, deque from fastapi import Request from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware class RateLimitMiddleware(BaseHTTPMiddleware): def __init__(self, app, requests_per_minute: int = 60): super().__init__(app) self.requests_per_minute = requests_per_minute self.request_history = defaultdict(lambda: deque()) async def dispatch(self, request: Request, call_next): client_ip = request.client.host if request.client else "unknown" current_time = time.time() # Clean old requests history = self.request_history[client_ip] while history and current_time - history[0] > 60: history.popleft() # Check rate limit if len(history) >= self.requests_per_minute: return JSONResponse( status_code=429, content={"detail": "Rate limit exceeded"} ) history.append(current_time) return await call_next(request) ``` ```python Request Logging theme={null} """ Log all requests with timing and metadata """ import logging import time from fastapi import Request from starlette.middleware.base import BaseHTTPMiddleware class LoggingMiddleware(BaseHTTPMiddleware): def __init__(self, app, log_body: bool = False): super().__init__(app) self.log_body = log_body self.logger = logging.getLogger("request_logger") async def dispatch(self, request: Request, call_next): start_time = time.time() client_ip = request.client.host if request.client else "unknown" # Log request self.logger.info(f"Request: {request.method} {request.url.path} from {client_ip}") # Optionally log body if self.log_body and request.method in ["POST", "PUT", "PATCH"]: body = await request.body() if body: self.logger.info(f"Body: {body.decode()}") response = await call_next(request) # Log response duration = (time.time() - start_time) * 1000 self.logger.info(f"Response: {response.status_code} in {duration:.1f}ms") return response ``` ```python Security Headers theme={null} """ Add security headers to all responses """ from fastapi import Request from starlette.middleware.base import BaseHTTPMiddleware class SecurityHeadersMiddleware(BaseHTTPMiddleware): def __init__(self, app): super().__init__(app) async def dispatch(self, request: Request, call_next): response = await call_next(request) # Add security headers response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["X-XSS-Protection"] = "1; mode=block" response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" return response ``` ```python Request ID theme={null} """ Add unique request IDs for tracing """ import uuid from fastapi import Request from starlette.middleware.base import BaseHTTPMiddleware class RequestIDMiddleware(BaseHTTPMiddleware): def __init__(self, app): super().__init__(app) async def dispatch(self, request: Request, call_next): # Generate unique request ID request_id = str(uuid.uuid4()) # Store in request state request.state.request_id = request_id # Process request response = await call_next(request) # Add to response headers response.headers["X-Request-ID"] = request_id return response ``` ## Error Handling Handle exceptions in middleware: ```python theme={null} from fastapi import Request from fastapi.responses import JSONResponse from starlette.middleware.base import BaseHTTPMiddleware import logging logger = logging.getLogger(__name__) class ErrorHandlingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): try: response = await call_next(request) return response except Exception as e: # Log the error logger.error(f"Request failed: {e}") # Return error response as JSONResponse return JSONResponse( status_code=500, content={"detail": "Internal server error"} ) ``` Error responses must be returned as `JSONResponse` objects to ensure proper serialization and HTTP status codes. ## Adding Middleware to AgentOS ```python custom_middleware.py theme={null} from agno.os import AgentOS from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIChat db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") agent = Agent( name="Basic Agent", model=OpenAIChat(id="gpt-5-mini"), db=db, ) agent_os = AgentOS(agents=[agent]) app = agent_os.get_app() ``` ```python theme={null} # Add your custom middleware app.add_middleware( RateLimitMiddleware, requests_per_minute=100 ) app.add_middleware( LoggingMiddleware, log_body=False ) app.add_middleware(SecurityHeadersMiddleware) ``` ```python theme={null} if __name__ == "__main__": agent_os.serve(app="custom_middleware:app", reload=True) ``` ## Developer Resources Rate limiting and request logging middleware implementation. Custom FastAPI app with JWT middleware and AgentOS integration. Official FastAPI middleware documentation and examples. # JWT Middleware Source: https://docs.agno.com/agent-os/middleware/jwt Secure your AgentOS application with JWT token validation and RBAC v2.1.0 Authenticate and authorize requests to your AgentOS application using JWT tokens. The middleware extracts tokens from Authorization headers or cookies, validates them, and automatically injects user\_id, session\_id, and custom claims into your endpoints. The JWT middleware provides three main features: 1. **Token Validation**: Validates JWT tokens and handles authentication 2. **Parameter Injection**: Automatically injects user\_id, session\_id, and custom claims into endpoint parameters 3. **RBAC Authorization**: Validates scopes against required permissions for each endpoint ```python jwt_middleware_setup.py theme={null} from agno.os.middleware import JWTMiddleware from agno.os.middleware.jwt import TokenSource app.add_middleware( JWTMiddleware, verification_keys=["your-jwt-verification-key"], # or use JWT_VERIFICATION_KEY environment variable algorithm="RS256", # RS256 for asymmetric keys, HS256 for symmetric user_id_claim="sub", # Extract user_id from 'sub' claim session_id_claim="session_id", # Extract session_id from claim dependencies_claims=["name", "email", "roles"], # Additional claims validate=True, # Enable token validation authorization=True, # Enable RBAC scope checking verify_audience=True, # Verify `aud` claim matches AgentOS ID ) ``` ## Token Sources The middleware supports three token sources: Extract JWT from `Authorization: Bearer ` header. ```python theme={null} from agno.os.middleware.jwt import TokenSource app.add_middleware( JWTMiddleware, verification_keys=["your-key"], token_source=TokenSource.HEADER, # Default ) ``` Extract JWT from HTTP-only cookies for web applications. ```python theme={null} app.add_middleware( JWTMiddleware, verification_keys=["your-key"], token_source=TokenSource.COOKIE, cookie_name="access_token", # Default ) ``` Try both header and cookie (header takes precedence). ```python theme={null} app.add_middleware( JWTMiddleware, verification_keys=["your-key"], token_source=TokenSource.BOTH, cookie_name="access_token", # Default token_header_key="Authorization", # Default ) ``` ## JWKS File Support For environments using RSA keys managed via JWKS (JSON Web Key Set), you can point to a static JWKS file instead of providing raw public keys: ```python jwks_file_setup.py theme={null} app.add_middleware( JWTMiddleware, jwks_file="/path/to/jwks.json", algorithm="RS256", authorization=True, ) ``` The middleware will: 1. Load public keys from the JWKS file at startup 2. Match incoming tokens by their `kid` (key ID) header claim 3. Validate signatures using the appropriate key ### JWKS File Format The JWKS file should follow the standard format: ```json theme={null} { "keys": [ { "kty": "RSA", "kid": "my-key-id", "use": "sig", "alg": "RS256", "n": "0vx7agoebGc...", "e": "AQAB" } ] } ``` ### Environment Variable You can also set the JWKS file path via environment variable: ```bash theme={null} export JWT_JWKS_FILE="/path/to/jwks.json" ``` JWKS keys are tried first (matched by `kid`). If no matching key is found, the middleware falls back to `verification_keys` if provided. ## Parameter Injection The middleware automatically injects JWT claims into AgentOS endpoints. The following parameters are extracted from tokens and injected into requests: * `user_id` - User identifier from token claims * `session_id` - Session identifier from token claims * `dependencies` - Custom claims for agent tools * `session_state` - Custom claims for session management For example, the `/agents/{agent_id}/runs` endpoint automatically uses `user_id`, `session_id`, `dependencies`, and `session_state` from the JWT token when available. This is useful for: * Automatically using the `user_id` and `session_id` from your JWT token when running an agent * Automatically filtering sessions retrieved from `/sessions` endpoints by `user_id` (where applicable) * Automatically injecting `dependencies` from claims in your JWT token into the agent run, which then is available on tools called by your agent View the [full example](/agent-os/usage/middleware/jwt-middleware) for more details. ## Security Features Use strong verification keys, store them securely (not in code), and enable validation in production. **Token Validation**: When `validate=True`, the middleware: * Verifies JWT signature using the verification key * Checks token expiration (`exp` claim) * Returns 401 errors for invalid/expired tokens **Audience Verification**: When `verify_audience=True`, the middleware: * Validates the `aud` claim matches the AgentOS ID * Ensures tokens are only valid for the intended AgentOS instance * Returns 401 for tokens with mismatched audience **HTTP-Only Cookies**: When using cookies: * Set `httponly=True` to prevent JavaScript access (XSS protection) * Set `secure=True` for HTTPS-only transmission * Set `samesite="strict"` for CSRF protection ## RBAC Authorization Enable Role-Based Access Control (RBAC) to validate JWT scopes against required permissions: ```python jwt_with_rbac.py theme={null} app.add_middleware( JWTMiddleware, verification_keys=["your-jwt-key"], algorithm="RS256", authorization=True, # Enable RBAC verify_audience=True, # Verify aud matches AgentOS ID ) ``` When `authorization=True`, the middleware: * Checks the `scopes` claim in JWT tokens * Validates scopes against required permissions for each endpoint * Returns 403 Forbidden for insufficient permissions ### Scope Format | Format | Example | Description | | ---------------------- | --------------------- | ------------------------ | | `resource:action` | `agents:read` | Access all resources | | `resource::action` | `agents:my-agent:run` | Access specific resource | | `resource:*:action` | `agents:*:run` | Wildcard access | | `agent_os:admin` | `agent_os:admin` | Full admin access | ### Custom Scope Mappings Override or extend default scope mappings: ```python custom_scope_mappings.py theme={null} app.add_middleware( JWTMiddleware, verification_keys=["your-key"], authorization=True, scope_mappings={ # Override default "GET /agents": ["custom:agents:list"], # Add new endpoint "POST /custom/action": ["custom:write"], # Allow without scopes "GET /public": [], } ) ``` For detailed RBAC documentation including all available scopes and default mappings, see [RBAC Documentation](/agent-os/security/rbac). ## Excluded Routes Skip middleware for specific routes: ```python jwt_excluded_routes.py theme={null} app.add_middleware( JWTMiddleware, verification_keys=["your-key"], excluded_route_paths=[ "/health", "/auth/login", "/auth/register", "/public/*", # Wildcards supported ] ) ``` ## Configuration Options See the [JWTMiddleware Reference](/reference/agent-os/jwt-middleware) for the complete list of configuration options. ### Authentication Options | Parameter | Description | Default | | ------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------ | | `verification_keys` | List of keys for JWT verification. For RS256, use public keys. For HS256, use shared secrets. Each key is tried in order until one succeeds. | `JWT_VERIFICATION_KEY` env var | | `jwks_file` | Path to a static JWKS file containing public keys. Keys are matched by `kid` from the JWT header. | `JWT_JWKS_FILE` env var | | `secret_key` | **(Deprecated)** Use `verification_keys` instead. | - | | `algorithm` | JWT algorithm (RS256, HS256, ES256, etc.) | `"RS256"` | | `validate` | Enable token validation | `True` | ### Token Source Options | Parameter | Description | Default | | ------------------ | ---------------------------------------------------------- | -------------------- | | `token_source` | Where to extract token from: `HEADER`, `COOKIE`, or `BOTH` | `TokenSource.HEADER` | | `token_header_key` | Header key for Authorization (when using HEADER or BOTH) | `"Authorization"` | | `cookie_name` | Cookie name (when using COOKIE or BOTH) | `"access_token"` | ### Claim Extraction Options | Parameter | Description | Default | | ---------------------- | ------------------------------------------------------- | -------------- | | `user_id_claim` | JWT claim for user ID | `"sub"` | | `session_id_claim` | JWT claim for session ID | `"session_id"` | | `scopes_claim` | JWT claim for scopes | `"scopes"` | | `audience_claim` | JWT claim for audience/OS ID | `"aud"` | | `dependencies_claims` | List of claims to extract for `dependencies` parameter | `[]` | | `session_state_claims` | List of claims to extract for `session_state` parameter | `[]` | ### Authorization Options (RBAC) | Parameter | Description | Default | | ---------------------- | ----------------------------------------------------- | ------------------ | | `authorization` | Enable RBAC scope checking | `False` | | `verify_audience` | Verify `aud` claim matches AgentOS ID | `False` | | `scope_mappings` | Custom route-to-scope mappings (additive to defaults) | `None` | | `admin_scope` | Scope that grants full admin access | `"agent_os:admin"` | | `excluded_route_paths` | Routes to skip JWT/RBAC checks | See below | ### Examples JWT authentication using Authorization headers for API clients. JWT authentication using HTTP-only cookies for web applications. Custom FastAPI app with JWT middleware and AgentOS integration. Detailed RBAC scopes, permissions, and access control configuration. Complete JWT middleware class reference. ### External Resources Official PyJWT library documentation for JWT encoding and decoding. # AgentOS Middleware Source: https://docs.agno.com/agent-os/middleware/overview Add authentication, logging, monitoring, and security features to your AgentOS application using middleware v2.1.0 AgentOS is built on FastAPI, allowing you to add any [FastAPI/Starlette compatible middleware](https://fastapi.tiangolo.com/tutorial/middleware/) for authentication, logging, monitoring, and security. Agno provides built-in JWT middleware for authentication, and you can create custom middleware for rate limiting, request logging, and security headers. Additionally, Agno provides some built-in middleware for common use cases, including authentication. See the following guides: Create your own middleware for logging, rate limiting, monitoring, and security. Built-in JWT authentication with automatic parameter injection and claims extraction. Use the built-in JWT middleware with Role-based access control and fine-grained permission scopes. ## Quick Start Adding middleware to your AgentOS application is straightforward: ```python agent_os.py theme={null} from agno.os import AgentOS from agno.os.middleware import JWTMiddleware from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIChat from agno.agent import Agent db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") agent = Agent( name="Basic Agent", model=OpenAIChat(id="gpt-5-mini"), db=db, ) # Create your AgentOS app agent_os = AgentOS(agents=[agent]) app = agent_os.get_app() # Add middleware app.add_middleware( JWTMiddleware, verification_keys=["your-jwt-verification-key"], validate=True ) if __name__ == "__main__": agent_os.serve(app="agent_os:app", reload=True) ``` Test middleware thoroughly in your own staging environment before production deployment. **Performance Impact:** Each middleware layer adds latency to requests. ## Common Use Cases **Secure your AgentOS with JWT authentication:** * Extract tokens from headers or cookies * Automatic parameter injection (user\_id, session\_id) * Custom claims extraction for `dependencies` and `session_state` * Route exclusion for public endpoints [Learn more about JWT Middleware](/agent-os/middleware/jwt) **Control access with permission scopes:** * Validate JWT scopes against required permissions * Per-resource access control (specific agents/teams/workflows) * Admin scope for full access * Customizable scope mappings [Learn more about RBAC](/agent-os/security/rbac) **Prevent API abuse with rate limiting:** ```python theme={null} class RateLimitMiddleware(BaseHTTPMiddleware): def __init__(self, app, requests_per_minute: int = 60): super().__init__(app) self.requests_per_minute = requests_per_minute # ... implementation app.add_middleware(RateLimitMiddleware, requests_per_minute=100) ``` **Monitor requests and responses:** ```python theme={null} class LoggingMiddleware(BaseHTTPMiddleware): async def dispatch(self, request: Request, call_next): start_time = time.time() response = await call_next(request) process_time = time.time() - start_time # Log request details... return response ``` ## Middleware Execution Order Middleware is executed in reverse order of addition. The last middleware added runs first. ```python theme={null} app.add_middleware(MiddlewareA) # Runs third (closest to route) app.add_middleware(MiddlewareB) # Runs second app.add_middleware(MiddlewareC) # Runs first (outermost) # Request: C -> B -> A -> Your Route # Response: Your Route -> A -> B -> C ``` **Best Practice:** Add middleware in logical order: 1. **Security middleware first** (CORS, security headers) 2. **Authentication middleware** (JWT, session validation) 3. **Monitoring middleware** (logging, metrics) 4. **Business logic middleware** (rate limiting, custom logic) ## Developer Resources ### Examples JWT authentication using Authorization headers for API clients. JWT authentication using HTTP-only cookies for web applications. Rate limiting and request logging middleware implementation. Custom FastAPI app with JWT middleware and AgentOS integration. Detailed RBAC scopes, permissions, and access control. ### External Resources Official FastAPI middleware documentation and examples. Starlette middleware reference and implementation guides. # AgentOS Source: https://docs.agno.com/agent-os/overview Agno's production runtime and control plane for your agentic systems AgentOS is Agno's production runtime and control plane for deploying and managing your agentic systems at scale. Built on [FastAPI](https://fastapi.tiangolo.com/), it provides a robust, scalable infrastructure that you can run locally during development or deploy to your cloud environment for production workloads. AgentOS serves as the central hub for your AI agents, teams, and workflows, providing a unified API for execution, monitoring, and management. Whether you're running a single agent or orchestrating complex multi-agent systems, AgentOS gives you the tools to build production-ready agentic applications. With AgentOS, you can: * **Run Agents / Teams / Workflows**: Create new runs for your agents, teams and workflows, either with a new session or a existing one. * **Manage Sessions**: Retrieve, update and delete sessions. * **Manage Memories**: Retrieve, update and delete memories. * **Manage Knowledge**: Manage the content of your knowledge base. * **Manage Evals**: Retrieve, create, delete and update evals. This is the same API that powers the [AgentOS Control Plane](/agent-os/control-plane). However, you can also use these endpoints to power your own application! See the full [API reference](/reference-api/overview) for more details. ## Create your AgentOS Below is a basic example of how to create an AgentOS instance: ```python my_os.py theme={null} from agno.os import AgentOS agent_os = AgentOS( name="My AgentOS", description="My Multi-Agent Runtime", agents=[basic_agent], teams=[basic_team], workflows=[basic_workflow] ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="my_os:app", reload=True) ``` You can configure the behaviour of your AgentOS by passing the following parameters to the `AgentOS` class: * `agents`: List of agents to include in the AgentOS * `teams`: List of teams to include in the AgentOS * `workflows`: List of workflows to include in the AgentOS * `knowledge`: List of knowledge instances to include in the AgentOS * `interfaces`: List of interfaces to include in the AgentOS * See the [Interfaces](/agent-os/interfaces) section for more details. * `config`: Configuration file path or `AgentOSConfig` instance * See the [Configuration](/agent-os/config) page for more details. * `base_app`: Optional custom FastAPI app to use instead of creating a new one * See the [Custom FastAPI App](/agent-os/custom-fastapi/overview) page for more details. * `lifespan`: Optional lifespan context manager for the FastAPI app * See the [Lifespan](/agent-os/lifespan) page for more details. * `enable_mcp_server`: Turn your AgentOS into an MCP server * See the [MCP enabled AgentOS](/agent-os/mcp/mcp) page for more details. See the [AgentOS class reference](/reference/agent-os/agent-os) for more details. ## Learn more Learn how to authenticate with AgentOS Learn how to use the AgentOS API to run your Agent / Team / Workflow Learn how to configure your AgentOS to your needs # AgentOS Gateway Source: https://docs.agno.com/agent-os/remote-execution/gateway Create a unified API gateway for multiple AgentOS instances The gateway pattern allows you to create a single AgentOS instance that aggregates agents, teams, and workflows from multiple remote AgentOS instances. This provides a unified API endpoint for your distributed agentic infrastructure. ## Use Cases * **Unified API**: Single endpoint for all your agents across different servers * **Load distribution**: Spread specialized agents across multiple servers * **Microservices architecture**: Each service hosts its own agents * **Hybrid deployments**: Combine local and remote agents in one interface ## Basic Gateway Setup Create a gateway that aggregates remote agents from multiple servers: ```python theme={null} from agno.agent import RemoteAgent from agno.team import RemoteTeam from agno.workflow import RemoteWorkflow from agno.os import AgentOS # Create the gateway AgentOS gateway = AgentOS( id="api-gateway", description="Unified API gateway for distributed agents", agents=[ RemoteAgent(base_url="http://server-1:7778", agent_id="assistant-agent"), RemoteAgent(base_url="http://server-2:7778", agent_id="researcher-agent"), ], teams=[ RemoteTeam(base_url="http://server-3:7778", team_id="research-team"), ], workflows=[ RemoteWorkflow(base_url="http://server-4:7778", workflow_id="qa-workflow"), ], ) app = gateway.get_app() if __name__ == "__main__": gateway.serve(app="gateway:app", port=7777) ``` ## Combining Local and Remote Mix local agents with remote agents in the same gateway: ```python theme={null} from agno.agent import Agent, RemoteAgent from agno.models.openai import OpenAIChat from agno.os import AgentOS from agno.db.postgres import PostgresDb db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") # Local agent local_agent = Agent( name="Q&A Agent", id="question-answer-agent", model=OpenAIChat(id="gpt-5-mini"), instructions="You are a helpful question and answer assistant.", db=db, ) # Remote agents from other servers remote_assistant = RemoteAgent( base_url="http://assistant-server:7778", agent_id="assistant-agent", ) remote_researcher = RemoteAgent( base_url="http://research-server:7778", agent_id="researcher-agent", ) # Gateway combining both gateway = AgentOS( id="hybrid-gateway", agents=[local_agent, remote_assistant, remote_researcher], ) app = gateway.get_app() if __name__ == "__main__": gateway.serve(app="gateway:app", port=7777) ``` ## Complete Gateway Example Here's a complete example with agents, teams, workflows, and a local workflow that uses remote agents: ```python theme={null} from agno.agent import Agent, RemoteAgent from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIChat from agno.os import AgentOS from agno.team import RemoteTeam from agno.workflow import RemoteWorkflow, Workflow from agno.workflow.step import Step # Database for local components db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") # Local agents for a local workflow story_writer = Agent( name="Story Writer", model=OpenAIChat(id="gpt-4o-mini"), instructions="Write a 100 word story based on the given topic", ) story_editor = Agent( name="Story Editor", model=OpenAIChat(id="gpt-4o-mini"), instructions="Review and improve the story's grammar and flow", ) # Local workflow using local agents story_workflow = Workflow( name="Story Generation", id="story-workflow", description="Generate and edit stories", db=db, steps=[ Step(name="write_story", agent=story_writer), Step(name="edit_story", agent=story_editor), ], ) # Gateway combining local and remote components gateway = AgentOS( id="content-gateway", description="Gateway for content generation services", agents=[ # Remote agents from specialized servers RemoteAgent(base_url="http://server-1:7778", agent_id="assistant-agent"), RemoteAgent(base_url="http://server-1:7778", agent_id="researcher-agent"), ], teams=[ # Remote team RemoteTeam(base_url="http://server-1:7778", team_id="research-team"), ], workflows=[ # Remote workflow RemoteWorkflow(base_url="http://server-1:7778", workflow_id="qa-workflow"), # Local workflow story_workflow, ], ) app = gateway.get_app() if __name__ == "__main__": gateway.serve(app="gateway:app", reload=True, port=7777) ``` See the [full example](/agent-os/usage/remote-execution/gateway) for more information. ## Authentication Considerations If authorization is enabled on remote servers and all endpoints are protected, not all gateway functions will work correctly. Specifically, the following endpoints need to be unprotected on remote servers for the gateway to work: * `/config` * `/agents` * `/agents/{agent_id}` * `/teams` * `/teams/{team_id}` * `/workflows` * `/workflows/{workflow_id}` # Remote Execution Source: https://docs.agno.com/agent-os/remote-execution/overview Execute agents, teams, and workflows hosted on remote AgentOS instances Remote execution enables you to run agents, teams, and workflows that are hosted on remote AgentOS instances. This is useful for: * **Distributed architectures**: Run specialized agents on different servers * **Microservices**: Decompose your agentic system into independent services * **Gateway pattern**: Create a unified API for multiple AgentOS instances ## Core Components Low-level client for direct API access to any AgentOS endpoint Execute agents on remote AgentOS instances Execute teams on remote AgentOS instances Execute workflows on remote AgentOS instances ## Quick Start ### 1. Set Up a Remote AgentOS Server First, create and run an AgentOS instance that will host your agents: ```python theme={null} # server.py from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.os import AgentOS agent = Agent( name="Assistant", id="assistant-agent", model=OpenAIChat(id="gpt-5-mini"), instructions="You are a helpful assistant.", ) agent_os = AgentOS( id="remote-server", agents=[agent], ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="server:app", port=7778) ``` Run the server: ```bash theme={null} python server.py ``` ### 2. Connect and Execute Remotely Use `RemoteAgent` to execute the agent from another application: ```python theme={null} import asyncio from agno.agent import RemoteAgent async def main(): agent = RemoteAgent( base_url="http://remote-server:7778", # Use localhost for local testing agent_id="assistant-agent", ) response = await agent.arun("Hello, how are you?") print(response.content) asyncio.run(main()) ``` ### 3. Create an AgentOS Gateway Create a gateway that aggregates multiple AgentOS instances: ```python theme={null} from agno.agent import RemoteAgent from agno.team import RemoteTeam from agno.workflow import RemoteWorkflow from agno.os import AgentOS local_agent = Agent( name="Research Agent", id="research-agent", model=OpenAIChat(id="gpt-5-mini"), instructions="You are a research assistant.", ) gateway = AgentOS( id="api-gateway", agents=[ local_agent, RemoteAgent(base_url="http://remote-server:7778", agent_id="assistant-agent"), ], ) app = gateway.get_app() if __name__ == "__main__": gateway.serve(app="gateway:app", port=7777) ``` See [Gateway Pattern](/agent-os/remote-execution/gateway) for more details. ## Authentication When connecting to authenticated AgentOS instances, pass the `auth_token` parameter: ```python theme={null} response = await agent.arun( "Hello", auth_token="your-jwt-token", ) ``` For the `AgentOSClient`, use the `headers` parameter: ```python theme={null} from agno.client import AgentOSClient client = AgentOSClient(base_url="http://localhost:7777") config = await client.aget_config(headers={"Authorization": "Bearer your-token"}) ``` ## Error Handling Remote execution can fail due to network issues or server unavailability: ```python theme={null} from agno.exceptions import RemoteServerUnavailableError try: response = await agent.arun("Hello") except RemoteServerUnavailableError as e: print(f"Server unavailable: {e.message}") print(f"Base URL: {e.base_url}") ``` ## Learn More Detailed guide on using RemoteAgent Detailed guide on using RemoteTeam Detailed guide on using RemoteWorkflow Create a unified API gateway for multiple AgentOS instances # Remote Agent Source: https://docs.agno.com/agent-os/remote-execution/remote-agent Execute agents hosted on remote AgentOS instances `RemoteAgent` allows you to execute agents that are running on a remote AgentOS instance as if they were local agents. This enables distributed architectures where specialized agents run on different servers. ## Prerequisites You need a running AgentOS instance with at least one agent. See [Creating Your First OS](/agent-os/creating-your-first-os) to set one up. ## Basic Usage ```python theme={null} import asyncio from agno.agent import RemoteAgent async def main(): # Connect to a remote agent agent = RemoteAgent( base_url="http://remote-server:7778", # Use localhost for local testing agent_id="assistant-agent", ) # Run the agent response = await agent.arun("What is the capital of France?") print(response.content) asyncio.run(main()) ``` ## Streaming Responses Stream responses in real-time for a better user experience: ```python theme={null} from agno.agent import RemoteAgent agent = RemoteAgent( base_url="http://remote-server:7778", # Use localhost for local testing agent_id="assistant-agent", ) async for event in agent.arun( "Tell me a story about a brave knight", stream=True, ): if hasattr(event, "content") and event.content: print(event.content, end="", flush=True) ``` ## Configuration Access Access the remote agent's configuration: ```python theme={null} from agno.agent import RemoteAgent agent = RemoteAgent( base_url="http://localhost:7777", agent_id="assistant-agent", ) # Access cached properties print(f"Name: {agent.name}") print(f"Description: {agent.description}") print(f"Tools: {agent.tools}") # Get fresh configuration config = await agent.get_agent_config() print(f"Model: {config.model}") # Force refresh cache agent.refresh_config() ``` ## Authentication For authenticated AgentOS instances: ```python theme={null} from agno.agent import RemoteAgent agent = RemoteAgent( base_url="http://localhost:7777", agent_id="assistant-agent", ) response = await agent.arun( "Hello", auth_token="your-jwt-token", ) ``` ## Error Handling ```python theme={null} from agno.agent import RemoteAgent from agno.exceptions import RemoteServerUnavailableError agent = RemoteAgent( base_url="http://localhost:7777", agent_id="assistant-agent", ) try: response = await agent.arun("Hello") except RemoteServerUnavailableError as e: print(f"Cannot connect to server: {e.message}") # Handle fallback logic ``` ## Developer Resources * [RemoteAgent Reference](/reference/agents/remote-agent) for complete API documentation. * [RemoteAgent Example](/agent-os/usage/remote-execution/remote-agent) for a complete example. # Remote Team Source: https://docs.agno.com/agent-os/remote-execution/remote-team Execute teams hosted on remote AgentOS instances `RemoteTeam` allows you to execute teams that are running on a remote AgentOS instance. This enables you to leverage complex multi-agent teams without hosting them locally. ## Prerequisites You need a running AgentOS instance with at least one team configured. See [Creating Your First OS](/agent-os/creating-your-first-os) to set one up. ## Basic Usage ```python theme={null} import asyncio from agno.team import RemoteTeam async def main(): # Connect to a remote team team = RemoteTeam( base_url="http://remote-server:7778", # Use localhost for local testing team_id="research-team", ) # Run the team response = await team.arun("Research the latest trends in AI") print(response.content) asyncio.run(main()) ``` ## Streaming Responses Stream team responses in real-time: ```python theme={null} from agno.team import RemoteTeam from agno.run.team import RunContentEvent team = RemoteTeam( base_url="http://remote-server:7778", # Use localhost for local testing team_id="research-team", ) print("Team Response: ", end="", flush=True) async for event in team.arun( "Analyze the current state of quantum computing", stream=True, ): if isinstance(event, RunContentEvent): print(event.content, end="", flush=True) ``` ## Configuration Access Access the remote team's configuration: ```python theme={null} from agno.team import RemoteTeam team = RemoteTeam( base_url="http://localhost:7777", team_id="research-team", ) # Access cached properties print(f"Name: {team.name}") print(f"Description: {team.description}") print(f"Role: {team.role}") # Get fresh configuration config = await team.get_team_config() print(f"Members: {config.members}") # Force refresh cache team.refresh_config() ``` ## Using in Gateway Register remote teams in an AgentOS gateway: ```python theme={null} from agno.team import RemoteTeam from agno.os import AgentOS gateway = AgentOS( id="api-gateway", teams=[ RemoteTeam(base_url="http://server-1:7777", team_id="research-team"), RemoteTeam(base_url="http://server-2:7777", team_id="analysis-team"), ], ) gateway.serve(port=7777) ``` ## Authentication For authenticated AgentOS instances: ```python theme={null} from agno.team import RemoteTeam team = RemoteTeam( base_url="http://localhost:7777", team_id="research-team", ) response = await team.arun( "Research this topic", auth_token="your-jwt-token", ) ``` ## Error Handling ```python theme={null} from agno.team import RemoteTeam from agno.exceptions import RemoteServerUnavailableError team = RemoteTeam( base_url="http://localhost:7777", team_id="research-team", ) try: response = await team.arun("Hello") except RemoteServerUnavailableError as e: print(f"Cannot connect to server: {e.message}") # Handle fallback logic ``` ## Developer Resources * [RemoteTeam Reference](/reference/teams/remote-team) for complete API documentation. * [RemoteTeam Example](/agent-os/usage/remote-execution/remote-team) for a complete example. # Remote Workflow Source: https://docs.agno.com/agent-os/remote-execution/remote-workflow Execute workflows hosted on remote AgentOS instances `RemoteWorkflow` allows you to execute workflows that are running on a remote AgentOS instance. This enables you to leverage complex multi-step workflows without hosting them locally. ## Prerequisites You need a running AgentOS instance with at least one workflow configured. See [Creating Your First OS](/agent-os/creating-your-first-os) to set one up. ## Basic Usage ```python theme={null} import asyncio from agno.workflow import RemoteWorkflow async def main(): # Connect to a remote workflow workflow = RemoteWorkflow( base_url="http://remote-server:7778", # Use localhost for local testing workflow_id="qa-workflow", ) # Run the workflow response = await workflow.arun("What are the benefits of using Python?") print(response.content) print(f"Status: {response.status}") asyncio.run(main()) ``` ## Streaming Responses Stream workflow responses in real-time: ```python theme={null} from agno.workflow import RemoteWorkflow workflow = RemoteWorkflow( base_url="http://remote-server:7778", # Use localhost for local testing workflow_id="story-workflow", ) print("Workflow Response: ", end="", flush=True) async for event in workflow.arun( "Write a story about space exploration", stream=True, ): # Handle content from agent steps or workflow completion if event.event == "RunContent" and hasattr(event, "content"): print(event.content, end="", flush=True) elif event.event == "WorkflowAgentCompleted" and hasattr(event, "content"): if event.content: print(event.content, end="", flush=True) ``` ## Passing Additional Data Send additional structured data to the workflow: ```python theme={null} from agno.workflow import RemoteWorkflow workflow = RemoteWorkflow( base_url="http://remote-server:7778", # Use localhost for local testing workflow_id="analysis-workflow", ) response = await workflow.arun( "Analyze the data", additional_data={ "metrics": {"revenue": 1000000, "growth": 0.15}, "period": "Q4 2024", }, ) ``` ## Configuration Access Access the remote workflow's configuration: ```python theme={null} from agno.workflow import RemoteWorkflow workflow = RemoteWorkflow( base_url="http://localhost:7777", workflow_id="qa-workflow", ) # Access cached properties print(f"Name: {workflow.name}") print(f"Description: {workflow.description}") # Get fresh configuration config = await workflow.get_workflow_config() # Force refresh cache workflow.refresh_config() ``` ## Using in Gateway Register remote workflows in an AgentOS gateway: ```python theme={null} from agno.workflow import RemoteWorkflow from agno.os import AgentOS gateway = AgentOS( id="api-gateway", workflows=[ RemoteWorkflow(base_url="http://server-1:7777", workflow_id="qa-workflow"), RemoteWorkflow(base_url="http://server-2:7777", workflow_id="analysis-workflow"), ], ) gateway.serve(port=7777) ``` ## Authentication For authenticated AgentOS instances: ```python theme={null} from agno.workflow import RemoteWorkflow workflow = RemoteWorkflow( base_url="http://localhost:7777", workflow_id="qa-workflow", ) response = await workflow.arun( "Process this request", auth_token="your-jwt-token", ) ``` ## Error Handling ```python theme={null} from agno.workflow import RemoteWorkflow from agno.exceptions import RemoteServerUnavailableError workflow = RemoteWorkflow( base_url="http://localhost:7777", workflow_id="qa-workflow", ) try: response = await workflow.arun("Hello") except RemoteServerUnavailableError as e: print(f"Cannot connect to server: {e.message}") # Handle fallback logic ``` ## Reference For complete API documentation, see [RemoteWorkflow Reference](/reference/workflows/remote-workflow). # AgentOS Security Source: https://docs.agno.com/agent-os/security/overview Secure your AgentOS instance with Role-Based Access Control (RBAC) Securing your agentic system is critical and AgentOS comes with built-in support for 2 security mechanisms: * Basic Authentication -> Simple key validation for development or for use during the MVP stage of your product. Not recommended for production systems. * Role-Based Access Control (RBAC): JWT powered validation that ensures only users authorized to make a request, are able to. ## Basic Authentication Basic authentication provides simple bearer-token authentication for securing your AgentOS instance. This is also refered to as "Security Key Authentication". ### Configuration Set the `OS_SECURITY_KEY` environment variable: ```bash theme={null} export OS_SECURITY_KEY="OSK_...your_key..." ``` And run your AgentOS application: ```python agent_os.py theme={null} from agno.os import AgentOS agent_os = AgentOS( id="my-agent-os", agents=[my_agent], ) if __name__ == "__main__": agent_os.serve(app="agent_os:app", reload=True) ``` When `OS_SECURITY_KEY` is configured, requests without a valid `Authorization: Bearer ` header return `401 Unauthorized`. ## Role-Based Access Control (RBAC) AgentOS provides built-in Role-Based Access Control (RBAC) to secure your entire agentic system. RBAC validates the identity of the user making the request using [JWT tokens](https://www.jwt.io/introduction). Whether this user is coming from the AgentOS Control Plane or your own application, the JWT token is validated and the scopes are checked against the required permissions for each endpoint. Learn more about how identity and authorization flow in AgentOS RBAC in the [RBAC](/agent-os/security/rbac) documentation. ### Enabling RBAC Enable RBAC by setting `authorization=True` when initializing AgentOS: ```python rbac_enabled.py theme={null} from agno.os import AgentOS from agno.os.config import AuthorizationConfig agent_os = AgentOS( id="my-agent-os", agents=[my_agent], authorization=True, # Enable authorization with RBAC ) ``` ### Configuring the Verification Key JWT verification on AgentOS requires a verification key to check the signature of the JWT token. When authorization is enabled on the AgentOS Control Plane, all traffic to your AgentOS will have a JWT token attached. Agno generates a private/public key pair, and the private key is used to sign the token. The public key can then be used to verify the signature of the JWT token. The key will be generated when you enable authorization: **When connecting a new OS during initial setup:**