Chapter 23 · AgentOS: Publishing Agents as Services

Learning objectives: Use AgentOS to turn Agents / Teams / Workflows into FastAPI services in one step; master auto-generated REST endpoints, streaming calls, and authorization configuration.

23.1 What Is AgentOS

The code in previous chapters was all "run and the process exits" scripts. AgentOS is Agno's official runtime: it wraps your agent into a FastAPI application you own and self-host, providing execution APIs, persistent state, authorization, tracing, and ops endpoints.

import os
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.os import AgentOS

db = SqliteDb(db_file="tmp/agentos.db")

support_agent = Agent(
    id="support-agent",            # id becomes part of the URL; set explicitly
    name="Support Agent",
    model=OpenAIChat(
        id="deepseek-chat",
        api_key=os.getenv("DEEPSEEK_API_KEY"),
        base_url="https://api.deepseek.com/v1",
    ),
    db=db,
)

agent_os = AgentOS(
    id="product-agent-os",
    agents=[support_agent],
    teams=[],          # Can also register teams=[...], workflows=[...]
    workflows=[],
    db=db,             # Components without their own db inherit this default
)

app = agent_os.get_app()   # Get the native FastAPI instance

if __name__ == "__main__":
    agent_os.serve(app="agent_os:app", reload=True)   # Listens on port 7777 by default

Visit http://localhost:7777 after startup to see the endpoints the runtime exposes. get_app() returns a standard FastAPI application, so the entire Uvicorn / Gunicorn / Docker deployment playbook applies.

23.2 Auto-Generated REST Endpoints

AgentOS organizes execution endpoints by「component type + component id」:

Component	Execution endpoint
Agent	POST /agents/{agent_id}/runs
Team	POST /teams/{team_id}/runs
Workflow	POST /workflows/{workflow_id}/runs

Call the customer-service Agent from above with curl:

curl http://localhost:7777/agents/support-agent/runs \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "message=Where is my order?" \
  -d "user_id=customer-42" \
  -d "session_id=order-support-42" \
  -d "stream=false"

Key points:

• The response includes run_id and session_id; reusing the same session_id stitches multiple calls into the same conversation thread;
• With stream=true the endpoint returns events as an SSE stream — add -N to curl to see chunked output;
• In addition to message, you can pass dependencies, session_state, metadata, knowledge_filters, and even output_schema (a JSON Schema string) to inject runtime context.

23.3 Wrapping Calls with a Python Client

You won't hand-write curl in production code. Wrap a session-managing client with httpx:

import httpx


class AgentOSClient:
    def __init__(self, base_url: str, agent_id: str, api_key: str | None = None):
        self.base_url = base_url.rstrip("/")
        self.agent_id = agent_id
        headers = {"x-api-key": api_key} if api_key else {}
        self.http = httpx.Client(base_url=self.base_url, headers=headers, timeout=120)

    def info(self) -> dict:
        """Standard handshake: probe auth_mode and component counts first."""
        return self.http.get("/info").json()

    def run(self, message: str, session_id: str, user_id: str) -> dict:
        resp = self.http.post(
            f"/agents/{self.agent_id}/runs",
            data={
                "message": message,
                "session_id": session_id,   # Reuse to keep the same thread
                "user_id": user_id,
                "stream": "false",
            },
        )
        resp.raise_for_status()
        return resp.json()


client = AgentOSClient("http://localhost:7777", "support-agent")
print(client.info()["auth_mode"])
reply = client.run("Where is my order?", session_id="order-42", user_id="customer-42")
print(reply["content"])

Drop the client class into your project's clients.py; both frontend BFFs and backend services reuse it, so session management converges in one place.

23.4 Discovery and Self-Introspection: /info and /config

Clients should probe instance capabilities before calling:

curl http://localhost:7777/info

Key fields in the response:

Field	Meaning
auth_mode	Current auth mode: none / security_key / jwt
agent_count / team_count / workflow_count	Number of registered components
mcp.enabled / mcp.path	Whether MCP service is mounted and at which path
agno_version	Runtime version

GET /config returns richer info including component IDs, database ID, interface and domain config. Frontends and gateways should adapt dynamically based on /info, not hardcode assumptions.

23.5 Authorization: Don't Let Your Agent Run Naked

Production environments must enable authorization:

agent_os = AgentOS(
    id="product-agent-os",
    agents=[support_agent],
    db=db,
    authorization=True,   # After enabling, validates requests via security_key or JWT
)

Two modes after enabling:

• security_key: for simple scenarios use a static key (Bearer token) to protect the entire instance — configured via env var OS_SECURITY_KEY; clients carry this credential in the request header;
• JWT: set JWT_VERIFICATION_KEY or JWT_JWKS_FILE env vars to integrate with your own identity system; control which components each user/role can call.

# No code changes needed: security_key mode is fully driven by environment variables
# (read via pydantic-settings). Before starting the service:
#   export AUTHORIZATION_ENABLED=true    # or pass authorization=True at construction
#   export OS_SECURITY_KEY="your-random-long-key"   # used as Bearer token for validation
agent_os = AgentOS(
    id="product-agent-os",
    agents=[support_agent],
    db=db,
    authorization=True,
)

After configuration, verify with curl: a request without credentials to a protected endpoint returns 401; adding -H "Authorization: Bearer $OS_SECURITY_KEY" works normally. Note: if both JWT-related vars and OS_SECURITY_KEY are set, the runtime warns and prefers JWT.

Security baseline

auth_mode=none should only appear in local development. Any AgentOS instance reachable from the public internet must have authorization=True, plus HTTPS and rate limiting at the gateway layer — exposing the model endpoint directly is equivalent to handing your API key and data to the entire internet.

23.6 Runtime Capability Overview

AgentOS is designed as "add capabilities on demand"; each capability has its own switch:

Need	Config
Persistence / tracing with a default database	db=...
JWT authorization	authorization=True
Execution tracing storage	tracing=True (covered in next chapter)
Expose the instance as an MCP Server	mcp_server=True
Scheduled tasks	scheduler=True
Plug into an existing FastAPI app	base_app=existing_app

base_app= is the key to incremental migration: an existing FastAPI project keeps its business routes untouched and only mounts the /agents/* prefix, avoiding a full rewrite just to introduce Agno.

Chapter Summary

• AgentOS = official runtime that packages components into a self-hosted FastAPI app; get_app() + serve() gets the service up in two steps;
• Execution endpoints are uniform: POST /{component}/{id}/runs; session_id maintains conversation threads;
• Probe GET /info (including auth_mode) before calling — that's the client-standard handshake;
• Production must have authorization=True; tracing, mcp_server, scheduler, base_app are all single-switch capabilities.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. What is the correct REST endpoint to invoke a specific Agent in AgentOS?

2. How do you make two HTTP calls belong to the same conversation thread?

3. How should a client determine which auth mode an AgentOS instance uses?

4. An existing large FastAPI project wants to introduce AgentOS without rewriting routes. What is the best approach?

🛠️ Hands-On Exercises

1. Start the service from this chapter using a method other than uvicorn agent_os:app --port 7777 (e.g., Gunicorn with multiple workers), and send three messages with two different session_ids to verify session isolation.
2. Write a Python client function chat(message, session_id) that wraps the call to /agents/support-agent/runs and handles SSE streaming responses.
3. With authorization=True enabled, call /info and /agents/*/runs with and without credentials; record which endpoints are public and which are blocked.

Once the service is live, how do you see what's happening inside? Next chapter: Observability and Debugging.