# Approvals Source: https://docs.agno.com/agent-os/approvals/overview Manage approval workflows for agents and teams via the AgentOS Control Plane. Approve, reject, and audit tool executions that require human authorization directly from AgentOS. ```python theme={null} from agno.agent import Agent from agno.approval import approval from agno.db.postgres import PostgresDb from agno.models.openai import OpenAIChat from agno.os import AgentOS from agno.tools import tool db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") @approval @tool(requires_confirmation=True) def delete_user_data(user_id: str) -> str: """Permanently delete all data for a user. Requires admin approval.""" return f"All data for user {user_id} has been deleted." agent = Agent( id="data-manager", model=OpenAIChat(id="gpt-4o-mini"), tools=[delete_user_data], instructions=["You help users manage data operations."], db=db, ) app = AgentOS( agents=[agent], db=db, ).get_app() ``` ## Approval Flow When a user triggers a tool decorated with `@approval`, the run pauses and a pending record is persisted to the database. An admin resolves the request via the AgentOS Control Plane or the API, and the run can then be continued. ## Managing Approvals View and resolve pending approvals from the AgentOS Control Plane. Each entry shows the agent, tool, arguments, and requesting user. Approvals list in AgentOS Control Plane Review details, approve or reject, and track resolution history. Approval required in agent chat ## Approval Types | Type | Behavior | Use Case | | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------- | | `@approval` (default) | **Blocking.** Run pauses until an admin approves or rejects. | Deletions, payments, bulk operations | | `@approval(type="audit")` | **Non-blocking.** The run pauses only for the tool's HITL step. A resolved audit record is created after the step completes. Requires a HITL flag on `@tool()`, such as `requires_confirmation`. | Compliance logging, activity auditing | ## Approvals API | Operation | Endpoint | | ------------------- | --------------------------------------- | | List approvals | `GET /approvals` | | Get approval | `GET /approvals/{approval_id}` | | Get approval status | `GET /approvals/{approval_id}/status` | | Get approval count | `GET /approvals/count` | | Resolve approval | `POST /approvals/{approval_id}/resolve` | | Delete approval | `DELETE /approvals/{approval_id}` | When authorization is enabled, resolving requires the `approvals:write` scope. The same scope also lets you continue a run that is still paused on a required approval. ## Next Steps | Task | Guide | | ------------------------- | --------------------------------------------------------------------------------- | | Blocking approval basics | [Approval basic](/examples/agents/approvals/approval-basic) | | List and resolve workflow | [Approval list and resolve](/examples/agents/approvals/approval-list-and-resolve) | | Audit-style approvals | [Audit approval](/examples/agents/approvals/audit-approval-confirmation) | | Team-level approvals | [Team approval](/examples/agents/approvals/approval-team) | | API reference | [Approval API schemas](/reference-api/schema/approvals/list-approvals) | # 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. Guardrails are an exception: they always run synchronously, even when `run_hooks_in_background=True`, so their checks can still block the run. ### 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 * `session_state` - The current session state * `dependencies` - The dependencies passed to the run * `metadata` - The run metadata 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 * [@hook Decorator Reference](/reference/hooks/hook-decorator) * [Hooks Overview](/hooks/overview) # A2A Client Source: https://docs.agno.com/agent-os/client/a2a-client Connect to any A2A-compatible agent server The `A2AClient` provides a Python interface for communicating with any [A2A protocol](https://a2a-protocol.org/) compatible server. This includes: * Agno AgentOS instances with A2A interface enabled * Google ADK agents * Any other A2A-compatible agent server ## Quick Start ### Connecting to Agno AgentOS via A2A interface ```python theme={null} import asyncio from agno.client.a2a import A2AClient async def main(): # Connect to an Agno AgentOS A2A endpoint client = A2AClient("http://localhost:7003/a2a/agents/my-agent") # Send a message result = await client.send_message(message="Hello!") print(result.content) asyncio.run(main()) ``` ### Connecting to Google ADK Google ADK uses JSON-RPC mode: ```python theme={null} from agno.client.a2a import A2AClient client = A2AClient("http://localhost:8001/", protocol="json-rpc") result = await client.send_message(message="Hello!") ``` ## Streaming Responses Stream responses in real-time: ```python theme={null} from agno.client.a2a import A2AClient client = A2AClient("http://localhost:7003/a2a/agents/my-agent") async for event in client.stream_message(message="Tell me a story"): if event.is_content and event.content: print(event.content, end="", flush=True) ``` ## Authentication AgentOS instances running with `authorization=True` require a JWT on every A2A request. Pass it via `headers`: ```python theme={null} import os from agno.client.a2a import A2AClient client = A2AClient("https://my-agent-os.com/a2a/agents/my-agent") headers = {"Authorization": f"Bearer {os.environ['AGENT_OS_JWT']}"} result = await client.send_message(message="Hello!", headers=headers) async for event in client.stream_message(message="Hello!", headers=headers): ... ``` `send_message`, `stream_message`, and `get_agent_card` all accept `headers`. The token needs the target's run scope (`agents:run`, or per-resource `agents:my-agent:run`) for `message:send` and `message:stream`. See [Scopes](/agent-os/security/authorization/scopes) for the full mapping. The `user_id` parameter sets `userId` in the message metadata. AgentOS honors it for anonymous callers only. When the request carries a JWT, the run is attributed to the token's principal and `user_id` is ignored. ## Developer Resources * [A2A Protocol Documentation](https://a2a-protocol.org/) * [A2AClient Reference](/reference/clients/a2a-client) * [Agno A2A Interface](/agent-os/interfaces/a2a/introduction) # AgentOS Client Source: https://docs.agno.com/agent-os/client/agentos-client Connect to Agno AgentOS instances via REST API 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 Stream responses in real-time for a better user experience: ```python theme={null} from agno.client import AgentOSClient from agno.run.agent import RunContentEvent, RunCompletedEvent client = AgentOSClient(base_url="http://localhost:7777") 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.client import AgentOSClient from agno.exceptions import RemoteServerUnavailableError client = AgentOSClient(base_url="http://localhost:7777") try: config = await client.aget_config() except RemoteServerUnavailableError as e: print(f"Server unavailable: {e.message}") print(f"URL: {e.base_url}") ``` ## API Reference For complete method documentation, parameters, and response types, see the [AgentOSClient Reference](/reference/clients/agentos-client). ## 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 # Clients Source: https://docs.agno.com/agent-os/client/overview Python clients for connecting to agent servers Agno provides Python clients for programmatic access to agent servers. These clients enable you to run agents, manage sessions, and integrate AI capabilities into your applications. ## Available Clients Connect to Agno AgentOS instances via REST API. Run agents, teams, workflows, manage sessions, knowledge, and memories. Connect to any A2A-compatible server including Agno, Google ADK, and other A2A implementations. ## Choosing a Client | Client | Use Case | | --------------- | -------------------------------------------------------------------------------------------------- | | `AgentOSClient` | Connect to Agno AgentOS instances with full feature access (sessions, knowledge, memories, traces) | | `A2AClient` | Connect to any A2A-compatible server for cross-framework agent communication | ## Quick Comparison ### AgentOSClient Best for connecting to Agno AgentOS instances where you need full access to all features: ```python theme={null} import asyncio from agno.client import AgentOSClient async def main(): client = AgentOSClient(base_url="http://localhost:7777") config = await client.aget_config() result = await client.run_agent(agent_id="my-agent", message="Hello!") sessions = await client.get_sessions(user_id="user-123") asyncio.run(main()) ``` ### A2AClient Best for cross-framework communication or connecting to A2A-compatible servers: ```python theme={null} import asyncio from agno.client.a2a import A2AClient async def main(): # Connect to an Agno A2A endpoint client = A2AClient("http://localhost:7003/a2a/agents/my-agent") result = await client.send_message(message="Hello!") # Or connect to Google ADK client = A2AClient("http://localhost:8001/", protocol="json-rpc") result = await client.send_message(message="Hello!") asyncio.run(main()) ``` # AgentOS Configuration Source: https://docs.agno.com/agent-os/config Configure quick prompts, display names, and per-database settings with a YAML file or the AgentOSConfig class AgentOS configuration allows you to customize your instance for different environments and deployment scenarios. You can control which AI models are available globally, 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 evals: display_name: "Evaluations" dbs: - db_id: db-0001 domain_config: display_name: Production evals ``` 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", "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). # Connect Your AgentOS Source: https://docs.agno.com/agent-os/connect-your-os Connect an AgentOS runtime to the Control Plane to manage and monitor it from one web interface. Connect local, staging, and production AgentOS runtimes to the Control Plane. Test registered components, inspect sessions and traces, and manage runtime data from the same interface. The Control Plane connects directly from your browser to the runtime endpoint. 1. Open [os.agno.com](https://os.agno.com) and sign in 2. Click **Add new OS** ## Configure the connection | Field | Description | | ---------------- | -------------------------------------------------------- | | **Environment** | Local (`http://localhost:7777`) or Live (your HTTPS URL) | | **Endpoint URL** | Where your AgentOS is running | | **OS Name** | A descriptive name, e.g. "Development OS" | | **Tags** | Optional. Organize with labels like `dev`, `stg`, `prd` | Click **CONNECT**. If successful, your runtime appears in the dashboard. ## Verify the connection | Indicator | Expected | | ------------ | -------------------------------------------------------------- | | Status | The runtime shows as running | | Components | Registered agents, teams, and workflows appear | | Runtime data | Sessions, memory, knowledge, and traces appear when configured | # AgentOS Control Plane Source: https://docs.agno.com/agent-os/control-plane Manage and monitor AgentOS runtimes from one web interface. The AgentOS Control Plane at [os.agno.com](https://os.agno.com) lets you manage and monitor connected AgentOS runtimes. Chat with agents, inspect traces and sessions, manage knowledge and memory, review approvals, and operate schedules from one web interface. AgentOS Control Plane Dashboard The Control Plane connects directly from your browser to your AgentOS runtime. ## Chat Interface Chat with agents, collaborate with teams, and run workflows from one screen. See [Agents](/agents/overview), [Teams](/teams/overview), and [Workflows](/workflows/overview) for more details. ### Agents Select an agent from the right panel and start a conversation. Each agent maintains its own history, tools, and instructions. Switching agents won't mix contexts. ### Teams Switch the toggle to Teams. Use the chat stream to watch how the team divides and solves the task. ### Workflows Switch to Workflows. Provide input (plain text or structured, depending on the workflow) and watch execution live as steps stream, produce output, and finish. ## Studio Build agents, teams, and workflows visually on a live canvas. Drag, drop, configure, and deploy without leaving the control plane. See [Studio](/agent-os/studio/introduction) for setup details. ## Approvals Pending tool executions can require human confirmation. Approvals pause runs until an admin resolves the request. See [Approvals](/agent-os/approvals/overview) for setup details. ## Tracing Traces capture the complete execution flow of agents and teams. Each trace contains spans representing individual operations (LLM calls, tool executions, etc.) with token usage, latency, and error information. ### Tree View Displays spans in a hierarchical structure showing parent-child relationships between operations. Useful for understanding how teams delegate to agents and how agents invoke tools. ### Waterfall View Visualizes spans on a time axis showing when each operation started, how long it took, and which operations ran in parallel. Helps identify bottlenecks and optimize performance. See [Tracing](/agent-os/tracing/overview) for configuration and filter options. ## Session Tracking Sessions group related runs under a single `session_id`. Each session captures messages, tool calls, metrics, and summaries for a single conversation. Sessions are stored in your database and can be queried and managed from the control plane. See [Chat History](/database/chat-history) and [Session Storage](/database/session-storage) for more details. ## Knowledge Knowledge bases provide agents with domain-specific information through RAG (Retrieval-Augmented Generation). Manage multiple knowledge bases, add content from URLs, files, or text, and monitor embedding status. Each knowledge base stores documents, chunks them, generates embeddings, and makes them searchable for agents. See [Manage Knowledge](/agent-os/knowledge/manage-knowledge) for setup details. ## Memories Memories store information that agents learn about users across conversations. Each memory is tied to a user ID and contains content, topics, timestamps, and the input that generated it. You can view, create, edit, or delete memories from the control plane. See [Memory](/memory/overview) for more details. ## Schedules Automate agents, teams, and workflows via AgentOS cron schedules. Configure custom intervals, retry logic, and timezones. Use the control plane to monitor run history, enable/disable job states, or trigger manual executions. For setup and API details, see [Scheduler](/agent-os/scheduler/overview) documentation. ## Managing Your AgentOS Connect and inspect your AgentOS runtimes from a single interface. Switch between local development and production instances and monitor connection health. See [Connect Your AgentOS](/agent-os/connect-your-os) for the connection steps. ## Authorization Enable JWT verification and generate key pairs from the control plane, either when connecting an OS or later from Settings. See [AgentOS Security](/agent-os/security/overview) and [Authorization](/agent-os/security/authorization/overview) for setup details. ## User Management Invite team members by entering their email addresses. Separate multiple emails with commas or press Enter/Tab between addresses. Every organization has three default roles. Custom roles are available on the Enterprise plan. | Role | Access | | ----------------- | ------------------------------------------------------------------------------------------------ | | **Owner** | Full access including billing, organization settings, and deletion | | **Administrator** | Manage members, roles, settings, and resources; cannot update billing or delete the organization | | **Member** | Run and edit AgentOS resources; cannot delete resources or manage members | See [Roles](/agent-os/security/authorization/roles) for the full capability matrix and custom role setup. # Overriding Routes Source: https://docs.agno.com/agent-os/custom-fastapi/override-routes Resolve route conflicts between your custom FastAPI app and AgentOS with the on_route_conflict parameter. 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 The `on_route_conflict` parameter accepts three values: | Option | Custom Routes | AgentOS Routes | On Conflict | | ---------------------------- | ------------- | -------------------------- | ------------------- | | `preserve_base_app` | ✓ Preserved | ✗ Skipped (conflicts only) | Logs at debug level | | `preserve_agentos` (default) | ✗ Overridden | ✓ Preserved | Logs a warning | | `error` | N/A | N/A | Raises `ValueError` | Non-conflicting routes from both your app and AgentOS are always included in the two preserve modes. ## Example The example below uses `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.hackernews import HackerNewsTools 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-5"), db=db, tools=[HackerNewsTools()], 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 Integrate your own FastAPI app with AgentOS. AgentOS is built on FastAPI, which means you can integrate your existing FastAPI application, routes, middleware, dependencies, and deployment entrypoints with AgentOS. ```python app.py theme={null} from agno.agent import Agent from agno.models.openai import OpenAIResponses from agno.os import AgentOS from fastapi import FastAPI app = FastAPI(title="Product API") @app.get("/account/{account_id}") async def get_account(account_id: str): return {"account_id": account_id, "status": "active"} support_agent = Agent( id="support-agent", model=OpenAIResponses(id="gpt-5.4"), ) agent_os = AgentOS( agents=[support_agent], base_app=app, ) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="app:app", reload=True) ``` The returned app serves both routes: | Route | Owner | | --------------------------------- | ----------- | | `GET /account/{account_id}` | Product API | | `POST /agents/support-agent/runs` | AgentOS | ## Handle Route Conflicts Set `on_route_conflict` when the base application already defines a path and method used by AgentOS. | Value | Behavior | Use when | | --------------------- | ----------------------------------------------- | ------------------------------------------------- | | `"preserve_agentos"` | AgentOS replaces the conflicting base-app route | AgentOS should own its standard API paths | | `"preserve_base_app"` | AgentOS skips the conflicting route | The product route must keep its existing behavior | | `"error"` | Application construction raises a `ValueError` | Every conflict should block deployment | `"preserve_agentos"` is the default. ```python theme={null} agent_os = AgentOS( agents=[support_agent], base_app=app, on_route_conflict="error", ) ``` See [Override Routes](/agent-os/custom-fastapi/override-routes) for conflict examples and matching behavior. ## Keep Existing Application Behavior AgentOS prepares the supplied FastAPI app in place: | Existing behavior | AgentOS behavior | | -------------------- | ----------------------------------------------------------------------------------------- | | Routes and routers | Preserved unless they conflict with an AgentOS route | | Middleware | Retained on the combined application | | FastAPI dependencies | Continue to apply to the routes that declare them | | Lifespan | Combined with the lifespan passed to AgentOS | | CORS | Existing CORS middleware is updated with AgentOS origins, or AgentOS adds CORS middleware | Use AgentOS [authorization](/agent-os/security/authorization/overview) for runtime access control. Keep product-specific FastAPI dependencies on the custom routes that need them. ## Add Middleware Add FastAPI or Starlette middleware to the base app before calling `get_app()`: ```python theme={null} from starlette.middleware import Middleware from starlette.middleware.trustedhost import TrustedHostMiddleware app = FastAPI( middleware=[ Middleware( TrustedHostMiddleware, allowed_hosts=["api.example.com"], ) ] ) agent_os = AgentOS( agents=[support_agent], base_app=app, ) app = agent_os.get_app() ``` See [AgentOS Middleware](/agent-os/middleware/overview) for JWT validation, request context, logging, and custom middleware. ## Combine Lifespans Pass a lifespan to AgentOS when the runtime needs its own startup and shutdown work. AgentOS wraps the lifespan already configured on the base app. ```python theme={null} from contextlib import asynccontextmanager @asynccontextmanager async def agent_os_lifespan(app): app.state.runtime_status = "ready" yield app.state.runtime_status = "stopped" agent_os = AgentOS( agents=[support_agent], base_app=app, lifespan=agent_os_lifespan, ) app = agent_os.get_app() ``` See [Custom Lifespan](/agent-os/lifespan) for a complete example. ## Run the Combined App Install the AgentOS and FastAPI CLI dependencies: ```bash theme={null} uv pip install -U "agno[os]" openai "fastapi[standard]" ``` ```bash Development theme={null} fastapi dev app.py ``` ```bash Production theme={null} fastapi run app.py --host 0.0.0.0 --port 8000 ``` ```bash Uvicorn theme={null} uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4 ``` ## Next Steps | Task | Guide | | ----------------------------------------- | ----------------------------------------------------------- | | Resolve path and method conflicts | [Override Routes](/agent-os/custom-fastapi/override-routes) | | Add authentication and request middleware | [AgentOS Middleware](/agent-os/middleware/overview) | | Configure runtime authorization | [Authorization](/agent-os/security/authorization/overview) | | Run startup and shutdown logic | [Custom Lifespan](/agent-os/lifespan) | | Review `base_app` parameters | [AgentOS class reference](/reference/agent-os/agent-os) | # AgentFactory Source: https://docs.agno.com/agent-os/factories/agent-factory Build an Agent per request from verified middleware claims, client input, or any request-time context. v2.6.0 An `AgentFactory` is a registered callable that produces a fresh `Agent` for each request. Register it in `AgentOS(agents=[...])` alongside any prototype agents. ```python basic_factory.py theme={null} from agno.agent import Agent, AgentFactory from agno.db.postgres import PostgresDb from agno.factory import RequestContext from agno.models.openai import OpenAIResponses from agno.os import AgentOS db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") def build_tenant_agent(ctx: RequestContext) -> Agent: user_id = ctx.user_id or "anonymous" return Agent( model=OpenAIResponses(id="gpt-5.4"), db=db, instructions=f"You are a helpful assistant for tenant {user_id}. Be concise.", markdown=True, ) tenant_factory = AgentFactory( id="tenant-agent", db=db, factory=build_tenant_agent, name="Per-tenant assistant", description="Builds a personalized agent per tenant on each request.", ) agent_os = AgentOS(agents=[tenant_factory]) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="basic_factory:app", port=7777, reload=True) ``` Run it against the factory like any other agent: ```bash theme={null} curl -X POST http://localhost:7777/agents/tenant-agent/runs \ -F 'message=Hello, who are you?' \ -F 'user_id=tenant_42' \ -F 'stream=false' ``` See the [Factories reference](/reference/agent-os/factories) for the full constructor signature and parameter list. ## Async Factories Use an async callable when you need to fetch context from a database, an HTTP service, or any other awaitable resource. ```python theme={null} async def build_tenant_agent(ctx: RequestContext) -> Agent: profile = await fetch_tenant_profile(ctx.user_id) return Agent( model=OpenAIResponses(id="gpt-5.4"), db=db, instructions=profile["instructions"], ) tenant_factory = AgentFactory( id="tenant-agent", db=db, factory=build_tenant_agent, ) ``` AgentOS detects the coroutine and awaits it on every request. ## With an Input Schema Declare a Pydantic model on the factory to validate client-supplied `factory_input` before the factory runs. ```python input_schema_factory.py theme={null} from typing import Literal from pydantic import BaseModel from agno.agent import Agent, AgentFactory from agno.db.postgres import PostgresDb from agno.factory import RequestContext from agno.models.openai import OpenAIResponses from agno.os import AgentOS db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") PERSONAS = { "analyst": "You are a data-driven research analyst. Cite sources and use numbers.", "advisor": "You are a strategic advisor. Focus on actionable recommendations.", "skeptic": "You are a critical skeptic. Challenge assumptions and highlight risks.", } class ResearchInput(BaseModel): persona: Literal["analyst", "advisor", "skeptic"] = "analyst" depth: int = 3 def build_research_agent(ctx: RequestContext) -> Agent: cfg: ResearchInput = ctx.input return Agent( model=OpenAIResponses(id="gpt-5.4"), db=db, instructions=( f"{PERSONAS[cfg.persona]}\n\n" f"Research depth: {cfg.depth} (higher = more thorough)." ), markdown=True, ) research_factory = AgentFactory( id="research-agent", db=db, factory=build_research_agent, input_schema=ResearchInput, ) agent_os = AgentOS(agents=[research_factory]) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="input_schema_factory:app", port=7777, reload=True) ``` Send `factory_input` as a JSON string in the run request: ```bash theme={null} curl -X POST http://localhost:7777/agents/research-agent/runs \ -F 'message=What are the latest trends in AI?' \ -F 'factory_input={"persona": "skeptic", "depth": 5}' \ -F 'stream=false' ``` If `factory_input` does not validate, AgentOS returns 400 before the factory runs. If `factory_input` is omitted entirely, AgentOS validates `{}` against the schema: the request succeeds when every field has a default (like `ResearchInput` above), and 400s otherwise. Without an `input_schema`, omitted `factory_input` leaves `ctx.input` as `None`. ## Authorization From Verified Context Use `ctx.trusted.claims` and `ctx.trusted.scopes` for any decision that affects authorization. Trusted fields are populated by middleware that has verified the request, never by client input. ```python theme={null} def build_workspace_agent(ctx: RequestContext) -> Agent: role = ctx.trusted.claims.get("role") tools = [read_docs] if role in ("admin", "editor"): tools.append(write_docs) if role == "admin": tools.append(manage_members) return Agent(model=OpenAIResponses(id="gpt-5.4"), db=db, tools=tools) ``` The trust split keeps authorization decisions visible at code review time. See [RequestContext fields](/reference/agent-os/factories#requestcontext) for the full schema and [JWT Role Factory](/examples/agent-os/factories/jwt-role-factory) for an end-to-end example. ## Error Handling Raise `FactoryPermissionError` from inside the factory to reject unauthorized callers with HTTP 403. AgentOS raises `FactoryValidationError` (400) automatically when `factory_input` fails `input_schema` validation. ```python theme={null} from agno.factory import FactoryPermissionError def build_agent(ctx: RequestContext) -> Agent: if "agents:run" not in ctx.trusted.scopes: raise FactoryPermissionError("Missing 'agents:run' scope") ... ``` See the [Factories reference](/reference/agent-os/factories) for the full exception hierarchy and the post-resolve behavior. ## Developer Resources * [Factories overview](/agent-os/factories/overview) * [RequestContext fields](/reference/agent-os/factories#requestcontext) * [Factories reference](/reference/agent-os/factories) * [Factory examples](/examples/agent-os/factories/overview) # Dynamic Agents Source: https://docs.agno.com/agent-os/factories/overview Build agents, teams, and workflows per request from JWT claims, user input, and other request-time context. v2.6.0 Build a fresh Agent, Team, or Workflow for each incoming request. A factory is a callable that AgentOS invokes per request, so tools, instructions, model, and database scope can depend on who's calling. ```python tenant_factory.py theme={null} from agno.agent import Agent, AgentFactory from agno.db.postgres import PostgresDb from agno.factory import RequestContext from agno.models.openai import OpenAIResponses from agno.os import AgentOS db = PostgresDb( id="factory-demo-db", db_url="postgresql+psycopg://ai:ai@localhost:5532/ai", ) def build_tenant_agent(ctx: RequestContext) -> Agent: user_id = ctx.user_id or "anonymous" return Agent( model=OpenAIResponses(id="gpt-5.4"), db=db, instructions=f"You are a helpful assistant for tenant {user_id}. Be concise.", markdown=True, ) tenant_factory = AgentFactory( id="tenant-agent", db=db, factory=build_tenant_agent, name="Per-tenant assistant", description="Builds a personalized agent per tenant on each request.", ) agent_os = AgentOS(agents=[tenant_factory]) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="tenant_factory:app", port=7777) ``` Hit `POST /agents/tenant-agent/runs` and the factory is invoked with a [`RequestContext`](/reference/agent-os/factories#requestcontext) for that request. `id` and `db` are required on `AgentFactory`. The factory's `id` overrides any `id` set on the `Agent` returned by the callable. ## When to Use a Factory Use a plain `Agent` / `Team` / `Workflow` when the component is shared across all callers. Reach for a factory when construction depends on the request. | Use Case | Pattern | | ------------------------------------- | ------------------------------------------------------------ | | Tools or model vary per caller role | `AgentFactory` reading `ctx.trusted.claims` | | Members vary per tenant | `TeamFactory` returning a `Team` with tenant-specific agents | | Pipeline shape varies per request | `WorkflowFactory` returning a different step graph | | Client picks persona, depth, or style | Factory with `input_schema` reading `ctx.input` | `GET /agents/{id}`, `GET /teams/{id}`, and `GET /workflows/{id}` (the component-detail endpoints) return the factory's metadata without invoking it. See the [Factories reference](/reference/agent-os/factories) for per-endpoint behavior and discovery payload shape. ## Learn How To Produce a fresh Agent per request from verified claims or client input. Compose a fresh Team and its members for each request. Compose pipelines whose step graph depends on the request. Fields, sources, and identity precedence. Populate `ctx.trusted.claims` from verified JWTs. Class signatures, parameters, and exceptions. ## Developer Resources * [Factory examples](/examples/agent-os/factories/overview) * [AgentFactory / TeamFactory / WorkflowFactory reference](/reference/agent-os/factories) * [RequestContext reference](/reference/agent-os/factories#requestcontext) # TeamFactory Source: https://docs.agno.com/agent-os/factories/team-factory Build a Team per request whose members, model, and instructions depend on the caller. v2.6.0 A `TeamFactory` produces a fresh `Team` for each request. Register it in `AgentOS(teams=[...])` alongside any prototype teams. ```python basic_team_factory.py theme={null} from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.factory import RequestContext from agno.models.openai import OpenAIResponses from agno.os import AgentOS from agno.team import Team, TeamFactory db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") def build_support_team(ctx: RequestContext) -> Team: user_id = ctx.user_id or "anonymous" billing_agent = Agent( name="Billing Agent", role="Handle billing inquiries", model=OpenAIResponses(id="gpt-5.4"), instructions=f"You handle billing questions for tenant {user_id}. Be concise.", ) tech_agent = Agent( name="Tech Support Agent", role="Handle technical issues", model=OpenAIResponses(id="gpt-5.4"), instructions=f"You handle technical support for tenant {user_id}. Be concise.", ) return Team( name="Support Team", model=OpenAIResponses(id="gpt-5.4"), members=[billing_agent, tech_agent], db=db, instructions=[ f"You are the support team leader for tenant {user_id}.", "Route billing questions to the Billing Agent and technical issues to the Tech Support Agent.", ], markdown=True, ) support_team_factory = TeamFactory( id="support-team", db=db, factory=build_support_team, name="Per-tenant Support Team", description="Builds a support team with billing and tech agents per tenant.", ) agent_os = AgentOS(teams=[support_team_factory]) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="basic_team_factory:app", port=7777, reload=True) ``` Run the team like any other team: ```bash theme={null} curl -X POST http://localhost:7777/teams/support-team/runs \ -F 'message=I need help with billing and a technical issue' \ -F 'user_id=tenant_42' \ -F 'stream=false' ``` See the [Factories reference](/reference/agent-os/factories) for the full constructor signature and parameter list. ## Members That Scale with the Caller Member composition can change per request. A common pattern: scale team size and model quality with a subscription tier from `ctx.trusted.claims`. ```python theme={null} def build_research_team(ctx: RequestContext) -> Team: tier = ctx.trusted.claims.get("tier", "free") model_id = TIER_MODELS.get(tier, "gpt-4.1-mini") members = [researcher(model_id), writer(model_id)] if tier == "enterprise": members.append(reviewer(model_id)) return Team(name="Research Team", model=OpenAIResponses(id=model_id), members=members, db=db) ``` The `tier` claim comes from verified middleware, so the `tier` cannot be changed by the request body. See [Authorization From Verified Context](/agent-os/factories/agent-factory#authorization-from-verified-context) for the trust model and [JWT Role Factory](/examples/agent-os/factories/jwt-role-factory) for an end-to-end JWT example. ## Async Factories Use an async callable when you need to fetch context from a database, an HTTP service, or any other awaitable resource. ```python async_team_factory.py theme={null} async def build_team(ctx: RequestContext) -> Team: members = await load_members_for_tenant(ctx.user_id) return Team( name="Tenant Team", model=OpenAIResponses(id="gpt-5.4"), members=members, db=db, ) team_factory = TeamFactory(id="tenant-team", db=db, factory=build_team) ``` ## With an Input Schema Declare a Pydantic model on the factory to validate client-supplied `factory_input` before the factory runs. ```python team_factory_input.py theme={null} from pydantic import BaseModel class TeamConfig(BaseModel): include_reviewer: bool = False def build_team(ctx: RequestContext) -> Team: cfg: TeamConfig = ctx.input members = [researcher, writer] if cfg.include_reviewer: members.append(reviewer) return Team(name="Research", model=OpenAIResponses(id="gpt-5.4"), members=members, db=db) team_factory = TeamFactory( id="research-team", db=db, factory=build_team, input_schema=TeamConfig, ) ``` `AgentFactory` and `TeamFactory` instances cannot be passed inside `members`. Build per-member customization inside the team's factory. ## Error Handling Raise `FactoryPermissionError` from inside the factory to reject unauthorized callers with HTTP 403. AgentOS raises `FactoryValidationError` (400) automatically when `factory_input` fails `input_schema` validation. ```python theme={null} from agno.factory import FactoryPermissionError def build_team(ctx: RequestContext) -> Team: if "teams:run" not in ctx.trusted.scopes: raise FactoryPermissionError("Missing 'teams:run' scope") ... ``` See the [Factories reference](/reference/agent-os/factories) for the full exception hierarchy and the post-resolve behavior. ## Developer Resources * [Factories overview](/agent-os/factories/overview) * [RequestContext fields](/reference/agent-os/factories#requestcontext) * [Factories reference](/reference/agent-os/factories) * [Factory examples](/examples/agent-os/factories/overview) # WorkflowFactory Source: https://docs.agno.com/agent-os/factories/workflow-factory Build a Workflow per request whose steps, agents, and model depend on the caller. v2.6.0 A `WorkflowFactory` produces a fresh `Workflow` for each request. Register it in `AgentOS(workflows=[...])` alongside any prototype workflows. ```python basic_workflow_factory.py theme={null} from agno.agent import Agent from agno.db.postgres import PostgresDb from agno.factory import RequestContext from agno.models.openai import OpenAIResponses from agno.os import AgentOS from agno.workflow import Step, Workflow, WorkflowFactory db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai") def build_content_pipeline(ctx: RequestContext) -> Workflow: user_id = ctx.user_id or "anonymous" drafter = Agent( name="Drafter", model=OpenAIResponses(id="gpt-5.4"), instructions=( f"You are a content drafter for tenant {user_id}. " "Write a first draft based on the topic. Keep it focused and concise." ), ) editor = Agent( name="Editor", model=OpenAIResponses(id="gpt-5.4"), instructions=( f"You are an editor for tenant {user_id}. " "Review the draft for clarity, grammar, and structure. Output the final version." ), ) return Workflow( name="Content Pipeline", description="Draft then edit content", db=db, steps=[ Step(name="draft", description="Write the first draft", agent=drafter), Step(name="edit", description="Edit and finalize", agent=editor), ], ) content_pipeline_factory = WorkflowFactory( id="content-pipeline", db=db, factory=build_content_pipeline, name="Content Pipeline", description="Builds a draft-then-edit content workflow per tenant.", ) agent_os = AgentOS(workflows=[content_pipeline_factory]) app = agent_os.get_app() if __name__ == "__main__": agent_os.serve(app="basic_workflow_factory:app", port=7777, reload=True) ``` Run the workflow like any other workflow: ```bash theme={null} curl -X POST http://localhost:7777/workflows/content-pipeline/runs \ -F 'message=Write a blog post about sustainable energy' \ -F 'user_id=tenant_42' \ -F 'stream=false' ``` See the [Factories reference](/reference/agent-os/factories) for the full constructor signature and parameter list. ## Step Shape That Varies by Tier The step graph can change per request. A common pattern: add or drop steps based on a tier read from `ctx.trusted.claims`. ```python theme={null} def build_article_pipeline(ctx: RequestContext) -> Workflow: tier = ctx.trusted.claims.get("tier", "free") model_id = TIER_MODELS.get(tier, "gpt-4.1-mini") steps = [] if tier == "enterprise": steps.append(Step(name="research", agent=researcher(model_id))) steps.append(Step(name="draft", agent=drafter(model_id))) steps.append(Step(name="edit", agent=editor(model_id))) return Workflow(name="Article Pipeline", db=db, steps=steps) ``` The `tier` claim is read from verified middleware, so the `tier` cannot be changed by the request body. See [Authorization From Verified Context](/agent-os/factories/agent-factory#authorization-from-verified-context) for the trust model and [JWT Role Factory](/examples/agent-os/factories/jwt-role-factory) for an end-to-end JWT example. ## Async Factories Use an async callable when you need to fetch context from a database, an HTTP service, or any other awaitable resource. ```python async_workflow_factory.py theme={null} async def build_pipeline(ctx: RequestContext) -> Workflow: config = await fetch_pipeline_config(ctx.user_id) steps = [Step(name=s.name, agent=build_agent(s)) for s in config.steps] return Workflow(name="Pipeline", db=db, steps=steps) pipeline_factory = WorkflowFactory(id="pipeline", db=db, factory=build_pipeline) ``` ## With an Input Schema Declare a Pydantic model on the factory to validate client-supplied `factory_input` before the factory runs. ```python workflow_input_schema.py theme={null} from pydantic import BaseModel class PipelineConfig(BaseModel): include_research: bool = False def build_pipeline(ctx: RequestContext) -> Workflow: cfg: PipelineConfig = ctx.input steps = [] if cfg.include_research: steps.append(Step(name="research", agent=researcher)) steps.extend([ Step(name="draft", agent=drafter), Step(name="edit", agent=editor), ]) return Workflow(name="Pipeline", db=db, steps=steps) pipeline_factory = WorkflowFactory( id="pipeline", db=db, factory=build_pipeline, input_schema=PipelineConfig, ) ``` ## Error Handling Raise `FactoryPermissionError` from inside the factory to reject unauthorized callers with HTTP 403. AgentOS raises `FactoryValidationError` (400) automatically when `factory_input` fails `input_schema` validation. ```python theme={null} from agno.factory import FactoryPermissionError def build_pipeline(ctx: RequestContext) -> Workflow: if "workflows:run" not in ctx.trusted.scopes: raise FactoryPermissionError("Missing 'workflows:run' scope") ... ``` See the [Factories reference](/reference/agent-os/factories) for the full exception hierarchy and the post-resolve behavior. ## Factory Re-invocation on Read Endpoints `GET /workflows/{id}/runs/{run_id}` and `GET /workflows/{id}/runs` re-invoke the factory because the workflow has to be reconstructed to read its session state. Pass `factory_input` as a query parameter to drive the rebuild: ```bash theme={null} curl "http://localhost:7777/workflows/content-pipeline/runs/RUN_ID?session_id=SESSION_ID&factory_input=%7B%22tier%22%3A%22enterprise%22%7D" ``` ## Developer Resources * [Factories overview](/agent-os/factories/overview) * [RequestContext fields](/reference/agent-os/factories#requestcontext) * [Factories reference](/reference/agent-os/factories) * [Factory examples](/examples/agent-os/factories/overview) # A2A Source: https://docs.agno.com/agent-os/interfaces/a2a/introduction Expose Agno agents via the A2A protocol v2.1.2 The [Agent-to-Agent Protocol (A2A)](https://a2a-protocol.org/latest/topics/what-is-a2a/) is an open standard 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 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_agentos:app", reload=True) ``` By default, all local agents, teams, and workflows in the AgentOS are exposed via A2A. Remote instances (`RemoteAgent`, `RemoteTeam`, `RemoteWorkflow`) are excluded from this default. 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 agent, 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-team-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-team) 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-team). ### 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-workflow-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-workflow) 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-workflow). A2A clients expect a server to expose only a single agent. To use your Agno A2A interface with those clients, use `/a2a/agents/{id}/` (or `/a2a/teams/{id}/`, `/a2a/workflows/{id}/`) as the base URL. ## Authorization With [authorization](/agent-os/security/authorization/overview) enabled, A2A routes sit behind the same JWT verification and scope checks as the REST API. Send the token as `Authorization: Bearer `. | A2A method | Agents | Teams | Workflows | | --------------------------------- | ------------- | ------------ | ---------------- | | `.well-known/agent-card.json` | `agents:read` | `teams:read` | `workflows:read` | | `message:send` / `message:stream` | `agents:run` | `teams:run` | `workflows:run` | | `tasks:get` | `agents:read` | `teams:read` | Not exposed | | `tasks:cancel` | `agents:run` | `teams:run` | Not exposed | Per-resource scopes apply through the A2A prefix: a token with `agents:my-agent:run` can call `POST /a2a/agents/my-agent/v1/message:send`. The deprecated `/a2a/message/send` and `/a2a/message/stream` dispatch routes require `agents:run` at the route, then re-check the run scope for the resolved target's type (`agents:run`, `teams:run`, or `workflows:run`) before running it. See [Scopes](/agent-os/security/authorization/scopes) for the full route-to-scope mapping. ### Run identity For anonymous callers, the `X-User-ID` header and the `userId` field in `params.message.metadata` attribute the run to a user. Authenticated callers are pinned to their token's principal, and any client-supplied identity is ignored. Anonymous callers cannot claim reserved principals (`sa:*`, `__scheduler__`). ### Task ownership `tasks:get` requires `contextId` (the session the run belongs to) and returns `400` without it. For callers scoped to a user (a service account, or a JWT user under [user isolation](/agent-os/security/authorization/user-isolation)), `tasks:get` and `tasks:cancel` verify that the task belongs to that user in that session. `tasks:cancel` also requires `contextId` for these callers. Before v2.7, A2A routes were mounted outside the AgentOS auth middleware and accepted unauthenticated requests even with `authorization=True`. v2.7 moves A2A behind the unified auth layer. If your deployment relied on open A2A endpoints, mint credentials with the scopes above for those callers before upgrading. ## Connecting to A2A Servers Agno provides two ways to connect to A2A-compatible servers: ### Using A2AClient For direct A2A protocol access: ```python theme={null} from agno.client.a2a import A2AClient client = A2AClient("http://localhost:7003/a2a/agents/my-agent") result = await client.send_message( message="Hello!", headers={"Authorization": f"Bearer {token}"}, # Required when the server enforces authorization ) ``` See [A2A Client](/agent-os/client/a2a-client) for full documentation. ### Using RemoteAgent For a higher-level agent interface: ```python theme={null} from agno.agent import RemoteAgent agent = RemoteAgent( base_url="http://localhost:7003", agent_id="my-agent", protocol="a2a", ) response = await agent.arun("Hello!", auth_token=token) # auth_token required when the server enforces authorization ``` See [RemoteAgent Reference](/reference/agents/remote-agent) for full documentation. ## Developer Resources * [AgentOS Reference](/reference/agent-os/agent-os) * [A2A Client Documentation](/agent-os/client/a2a-client) * [A2A Protocol Documentation](https://a2a-protocol.org/latest/) * [Examples](/agent-os/usage/interfaces/a2a/basic) # 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](/other/v2-migration#8-apps-interfaces) for complete steps. ## Example usage ```bash theme={null} uv pip install 'agno[os,agui]' openai ``` 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 OpenAIResponses from agno.os import AgentOS from agno.os.interfaces.agui import AGUI chat_agent = Agent(model=OpenAIResponses(id="gpt-5.4")) 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 from the repository root: `pnpm install` 3. Build Dojo and its dependencies (including the Agno package): `pnpm build --filter=demo-viewer` 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/05_agent_os/16_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 typing import Optional from agno.run.agent import CustomEvent @dataclass class CustomerProfileEvent(CustomEvent): customer_name: Optional[str] = None customer_email: Optional[str] = None ``` **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"], ) yield f"Profile retrieved for {customer['name']}" ``` Custom events are streamed in real-time to the AG-UI frontend. See [Custom Events documentation](/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[Union[Agent, RemoteAgent]]` | `None` | Agno `Agent` or `RemoteAgent` instance. | | `team` | `Optional[Union[Team, RemoteTeam]]` | `None` | Agno `Team` or `RemoteTeam` instance. | | `prefix` | `str` | `""` | Route prefix (e.g., `/chat`, `/web-research`). | | `tags` | `Optional[List[str]]` | `["AGUI"]` | OpenAPI tags for the router. | Provide `agent` or `team`. ### Key Method | Method | Parameters | Return Type | Description | | ------------ | ---------- | ----------- | -------------------------------------------------------- | | `get_router` | None | `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. ## Authorization With [authorization](/agent-os/security/authorization/overview) enabled, `POST /agui` requires the run scope for the interface's bound entity: `agents:run` for an agent, `teams:run` for a team. Send the JWT as `Authorization: Bearer `. `GET /status` requires a valid token but no scope. For anonymous callers, `forwardedProps.user_id` attributes the run to a user. Authenticated callers are pinned to their token's principal, and a client-supplied `user_id` is ignored. See [Scopes](/agent-os/security/authorization/scopes) for the full route-to-scope mapping. ## 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. Override with `AGENT_OS_HOST` env var. | | `port` | `int` | `7777` | Port to bind. Override with `AGENT_OS_PORT` env var. | | `reload` | `bool` | `False` | Enable auto-reload for development. | | `reload_includes` | `Optional[List[str]]` | `None` | File patterns to watch for auto-reload. | | `reload_excludes` | `Optional[List[str]]` | `None` | File patterns to exclude from reload. | | `workers` | `Optional[int]` | `None` | Number of Uvicorn worker processes. | | `access_log` | `bool` | `False` | Enable Uvicorn access logging. | See [cookbook examples](https://github.com/agno-agi/agno/tree/main/cookbook/05_agent_os/16_agui) for updated interface patterns. # Discord Source: https://docs.agno.com/agent-os/interfaces/discord/introduction Deploy agents to Discord for community support and moderation. The Discord integration connects any Agno agent to Discord via the Gateway API, so you don't need to set up webhooks. ## Quick Start ```python discord_agent.py theme={null} from agno.agent import Agent from agno.models.openai import OpenAIResponses from agno.integrations.discord import DiscordClient agent = Agent(name="Assistant", model=OpenAIResponses(id="gpt-5.4")) discord_client = DiscordClient(agent) if __name__ == "__main__": discord_client.serve() ``` ```bash theme={null} uv pip install agno discord.py openai python discord_agent.py ``` ## How It Works | Concept | Behavior | | --------------- | ------------------------------------------------------------ | | **Gateway** | Direct connection to Discord, no webhooks or tunnels needed | | **Threads** | Auto-creates threads for conversations with isolated context | | **Intents** | Requires Message Content Intent enabled in Developer Portal | | **Permissions** | Send Messages, Read History, Create Threads, Attach Files | ## Setup You need a Discord Application with a bot user. Create one in the [Discord Developer Portal](https://discord.com/developers/applications). Then set your bot token as an environment variable: ```bash theme={null} export DISCORD_BOT_TOKEN="..." ``` ## Next Steps Bot setup, threads, intents, and permissions. Basic agent, media handling, and user memory. Give agents channel, message, and reaction capabilities. Create and manage your Discord application. # Interfaces Source: https://docs.agno.com/agent-os/interfaces/overview Expose agents, teams, and workflows through AI frontends, messaging platforms, and agent protocols. Interfaces expose AgentOS applications through AI frontends, messaging platforms, and agent protocols. AG-UI, Slack, Telegram, WhatsApp, and A2A are available as interfaces. ```python slack_agent.py theme={null} from agno.agent import Agent from agno.models.openai import OpenAIResponses from agno.os import AgentOS from agno.os.interfaces.slack import Slack support_agent = Agent( id="support-agent", model=OpenAIResponses(id="gpt-5.4"), ) agent_os = AgentOS( agents=[support_agent], interfaces=[Slack(agent=support_agent)], ) app = agent_os.get_app() ``` The Slack interface mounts its webhook routes on the AgentOS application and routes each conversation to `support_agent`. ## Choose an Interface | Interface | Use when | Components | | ------------------------------------------------------ | -------------------------------------------------------------------- | ------------------------ | | [AG-UI](/agent-os/interfaces/ag-ui/introduction) | A web or mobile frontend speaks the Agent-User Interaction Protocol | Agents, teams | | [Slack](/agent-os/interfaces/slack/introduction) | Teams use an agent in channels, direct messages, and threads | Agents, teams, workflows | | [Telegram](/agent-os/interfaces/telegram/introduction) | Users reach an agent through direct messages or group chats | Agents, teams, workflows | | [WhatsApp](/agent-os/interfaces/whatsapp/introduction) | Customers interact with an agent through WhatsApp Business | Agents, teams, workflows | | [A2A](/agent-os/interfaces/a2a/introduction) | Other agent systems call AgentOS through the Agent-to-Agent Protocol | Agents, teams, workflows | Discord uses the standalone `DiscordClient` integration from `agno.integrations.discord`. See the [Discord guide](/agent-os/interfaces/discord/introduction). ## How Interfaces Work Each interface mounts a FastAPI router on AgentOS and translates between the external protocol and an Agent, Team, or Workflow. | Responsibility | Behavior | | ---------------- | ----------------------------------------------------------------------------------------- | | Request handling | Parse platform events or protocol messages into Agno run input | | Session routing | Map the external user and conversation to AgentOS user and session IDs | | Streaming | Translate run events into the format supported by the client | | Media | Pass supported files, images, audio, and video to the component | | Responses | Return text, generated media, progress, and human-in-the-loop state through the interface | Support varies by interface. Use each interface guide for its events, media types, and response behavior. ## Authentication Messaging platforms authenticate their own webhook routes. Protocol interfaces follow the AgentOS authentication mode and scope mappings. | Interface | Request verification | | --------- | ----------------------------------------------------------------- | | Slack | Slack signing secret | | Telegram | Telegram webhook secret token | | WhatsApp | Meta webhook signature using the WhatsApp app secret | | AG-UI | AgentOS bearer authentication and run scopes when configured | | A2A | AgentOS bearer authentication and resource scopes when configured | The regular AgentOS REST routes continue to use the runtime's configured [authentication and authorization](/agent-os/security/overview). ## Mount Multiple Interfaces One AgentOS application can expose the same component through several interfaces: ```python theme={null} from agno.os.interfaces.agui import AGUI from agno.os.interfaces.slack import Slack agent_os = AgentOS( agents=[support_agent], interfaces=[ AGUI(agent=support_agent), Slack(agent=support_agent), ], ) app = agent_os.get_app() ``` The agent keeps one implementation while each interface owns protocol handling, routing, and response formatting. ## Next Steps | Task | Guide | | ------------------------------------ | ------------------------------------------------------ | | Connect a product frontend | [AG-UI](/agent-os/interfaces/ag-ui/introduction) | | Build a workplace agent | [Slack](/agent-os/interfaces/slack/introduction) | | Serve a messaging bot | [Telegram](/agent-os/interfaces/telegram/introduction) | | Reach customers on WhatsApp | [WhatsApp](/agent-os/interfaces/whatsapp/introduction) | | Expose agents to other agent systems | [A2A](/agent-os/interfaces/a2a/introduction) | # Features Source: https://docs.agno.com/agent-os/interfaces/slack/features Memory, files, streaming, search, and cross-platform identity for your Slack agent. A basic agent receives a message, thinks, and replies. A production agent handles the messy reality of workplace chat: conversations that span days, files dropped mid-thread, long-running tasks that need visibility, and sensitive actions that need approval. Here's what makes it production-ready: | Capability | What it does | Quick setup | | ----------------------------------------------- | ------------------------------------------ | ------------------------------------------ | | [Memory](#memory) | Conversations persist across messages | `db=SqliteDb(...)` | | [Files](#files) | Read attachments, send files back | `SlackTools(enable_upload_file=True)` | | [Streaming](#streaming) | Show progress on long tasks | On by default | | [Search](#search) | Find messages and context across workspace | `SlackTools(enable_search_workspace=True)` | | [Identity](#identity) | Recognize users across platforms | `resolve_user_identity=True` | | [Response control](#response-control) | Control when bot responds | `reply_to_mentions_only=True` | | [Approvals](/agent-os/interfaces/slack/hitl) | Pause for human approval | `@tool(requires_confirmation=True)` | | [Context provider](/context-providers/overview) | Query and update Slack from any agent | `SlackContextProvider()` | *** ## Memory Slack conversations span hours or days. With a database configured, the agent remembers previous messages in the thread and can reference earlier context without the user restating it. ```python theme={null} from agno.agent import Agent from agno.db.sqlite import SqliteDb agent = Agent( name="Support Bot", model=..., db=SqliteDb(db_file="sessions.db"), add_history_to_context=True, num_history_runs=5, ) ``` Each Slack thread maps to one session. DMs work the same way. All responses are sent as thread replies, keeping channel conversations organized. Conversations persist across server restarts. Session format: `{entity_id}:{thread_ts}`. For example, an agent named "Support Bot" gets the auto-generated ID `support-bot` and produces session IDs like `support-bot:1719000000.000100`. ## Files The agent can read files that users attach and generate files in response. The interface extracts attachments from incoming messages and makes them available to the agent. Generated files upload back to the thread. ```python theme={null} from agno.tools.slack import SlackTools agent = Agent( tools=[SlackTools(enable_upload_file=True, enable_download_file=True)], ) ``` ## Streaming Responses stream live, so users see text appear as the agent generates it, with task cards showing progress on long-running work. ```python theme={null} Slack(agent=agent, streaming=True) # on by default ``` Streaming requires [Agents & AI Apps](https://docs.slack.dev/ai) mode. Enable it under **Agents & AI Apps** in your [Slack App settings](https://api.slack.com/apps). Without it, users see a spinner until the full response is ready. ## Search The agent can search across your Slack workspace: messages, channels, and threads. ```python theme={null} from agno.tools.slack import SlackTools agent = Agent( tools=[SlackTools(enable_search_workspace=True)], ) ``` The `search_workspace` tool uses Slack's semantic search to find relevant messages and context. Useful for catching up on discussions, finding prior decisions, or summarizing activity. Search requires additional [OAuth scopes](https://docs.slack.dev/reference/scopes) (`search:read.public`, `search:read.files`, `search:read.users`). Add them under **OAuth & Permissions** in your [Slack App settings](https://api.slack.com/apps). See the [Reference](/agent-os/interfaces/slack/reference#oauth-scopes) for the full list. ## Identity The same user messages from Slack today, WhatsApp tomorrow. With identity resolution, the agent recognizes them as the same person. ```python theme={null} Slack(agent=agent, resolve_user_identity=True) ``` The agent receives `user_id` as the user's email instead of their Slack ID. Memory and context carry across platforms. ## Response control By default, the bot only responds when @mentioned in channels. In DMs, it responds to every message. | Setting | Behavior | | --------------------------------------- | -------------------------------------------- | | `reply_to_mentions_only=True` (default) | Responds to @mentions in channels, all DMs | | `reply_to_mentions_only=False` | Responds to every message in joined channels | Setting `reply_to_mentions_only=False` means the bot responds to every channel message. Use this only for dedicated bot channels. ## Troubleshooting **Check:** Server running? ngrok active? Request URL matches `{your-url}/slack/events`? **Fix:** Verify you've subscribed to `app_mention` and `message.im` events in your Slack App. **Cause:** Wrong signing secret, or request timestamp too old. **Fix:** `SLACK_SIGNING_SECRET` must match **Basic Information > App Credentials** in your Slack App settings. If using ngrok, restart it to clear stale connections. Signature verification rejects requests older than 5 minutes. **Cause:** "Agents & AI Apps" not enabled. **Fix:** In Slack App settings → **Agents & AI Apps** → toggle the feature **On**. **Cause:** `reply_to_mentions_only=False` makes the bot respond to every message in channels it can see, including @mentions of other users. **Fix:** Set `reply_to_mentions_only=True` (the default). Only set it to `False` for dedicated bot channels. **Cause:** The `search_workspace` tool requires an `action_token` that only exists when running through the Slack interface. It doesn't work in standalone scripts or terminal testing. **Fix:** Test search through the actual Slack bot, not in a standalone script. For terminal testing, use `search_messages` instead (requires a user token with `search:read` scope). **Cause:** System Python missing root certificates. **Fix:** ```bash theme={null} export SSL_CERT_FILE=$(python3 -c "import certifi; print(certifi.where())") ``` ## Next steps Pause for approval before sensitive actions All parameters and endpoints # Human-in-the-Loop Source: https://docs.agno.com/agent-os/interfaces/slack/hitl Pause agents for approval before executing sensitive tools. [Human-in-the-Loop](/hitl/overview) solves three problems that appear when agents move from answering questions to taking actions: 1. **Irreversible operations.** Sending an email, deleting a record, or posting to a channel cannot be undone. A human checkpoint prevents mistakes that require cleanup or apologies. 2. **Missing context.** The agent knows what action to take but lacks a critical detail. A deployment needs a target environment. A booking needs a budget. Rather than guessing, the agent pauses and asks. 3. **Audit trail.** Sensitive operations need accountability. Slack threads already contain the discussion that led to the action. Rendering the approval in the same thread keeps the decision and its context together.