第 22 章 · AgentOS：把 Agent 发布成服务
本章目标：用 AgentOS 把 Agent/Team/Workflow 一键变成 FastAPI 服务，掌握自动生成的 REST 接口、流式调用与鉴权配置。
22.1 AgentOS 是什么
前面章节的代码都是"跑完进程就结束"的脚本。AgentOS 是 Agno 官方的运行时（runtime）：它把你的智能体包装成一个你拥有并自行托管的 FastAPI 应用，提供执行 API、持久化状态、授权、追踪和运维端点。
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 将成为 URL 的一部分，务必显式设置
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=[], # 也可以注册 teams=[...], workflows=[...]
workflows=[],
db=db, # 组件未单独配库时继承该默认库
)
app = agent_os.get_app() # 拿到原生 FastAPI 实例
if __name__ == "__main__":
agent_os.serve(app="agent_os:app", reload=True) # 默认监听 7777 端口启动后访问 http://localhost:7777 即可看到运行时提供的接口。get_app() 返回的就是标准 FastAPI 应用，所以 Uvicorn/Gunicorn/Docker 的整套部署经验都适用。
22.2 自动生成的 REST 接口
AgentOS 按「组件类型 + 组件 id」组织执行端点：
| 组件 | 执行端点 |
|---|---|
| Agent | POST /agents/{agent_id}/runs |
| Team | POST /teams/{team_id}/runs |
| Workflow | POST /workflows/{workflow_id}/runs |
用 curl 调用刚才的客服 Agent：
curl http://localhost:7777/agents/support-agent/runs \
-H "Content-Type: application/x-www-form-urlencoded" \
-d "message=我的订单到哪了？" \
-d "user_id=customer-42" \
-d "session_id=order-support-42" \
-d "stream=false"要点：
- 响应中包含
run_id与session_id；复用同一个session_id就把多次调用串成同一会话线程； stream=true时端点以 SSE 流式返回事件，curl 加-N参数即可看到逐块输出；- 除了 message，还可以传
dependencies、session_state、metadata、knowledge_filters甚至output_schema（JSON Schema 字符串）来注入运行时上下文。
22.3 用 Python 客户端封装调用
生产代码里当然不会手写 curl。用 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:
"""标准握手：先探测 auth_mode 与组件数量"""
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, # 复用即同一线程
"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("我的订单到哪了？", session_id="order-42", user_id="customer-42")
print(reply["content"])把客户端类放进你项目的 clients.py，前端 BFF 或后端服务都复用它，会话管理就收敛到了一处。
22.4 发现与自省：/info 与 /config
客户端调用前应先探测实例能力：
curl http://localhost:7777/info返回的关键字段：
| 字段 | 含义 |
|---|---|
auth_mode | 当前鉴权模式：none / security_key / jwt |
agent_count / team_count / workflow_count | 注册的组件数量 |
mcp.enabled / mcp.path | MCP 服务是否挂载及路径 |
agno_version | 运行时版本 |
GET /config 则返回组件 ID、数据库 ID、接口与域配置等更完整的信息。前端或网关应基于 /info 动态适配，而不是硬编码假设。
22.4 鉴权：别让智能体裸奔
生产环境必须开启授权：
agent_os = AgentOS(
id="product-agent-os",
agents=[support_agent],
db=db,
authorization=True, # 开启后按 security_key 或 JWT 模式校验请求
)开启后的两种模式：
- security_key：简单场景下用静态密钥（Bearer token）保护整个实例——通过环境变量
OS_SECURITY_KEY配置，客户端在请求头携带该凭证； - JWT：设置
JWT_VERIFICATION_KEY或JWT_JWKS_FILE等环境变量接入你自己的身份体系，按用户/角色控制可调用的组件。
# 无需改代码：security_key 模式完全由环境变量驱动（pydantic-settings 读取）
# 启动服务前导出：
# export AUTHORIZATION_ENABLED=true # 或构造时传 authorization=True
# export OS_SECURITY_KEY="你的随机长密钥" # 作为 Bearer token 校验
agent_os = AgentOS(
id="product-agent-os",
agents=[support_agent],
db=db,
authorization=True,
)配置后可用 curl 验证：不带凭证请求受保护端点会得到 401，加上 -H "Authorization: Bearer $OS_SECURITY_KEY" 则正常返回。注意若同时设置了 JWT 相关变量和 OS_SECURITY_KEY，运行时会警告并以 JWT 为准。
安全底线
auth_mode=none 只能出现在本机开发。任何公网可达的 AgentOS 实例都要开 authorization=True，并在网关层叠加 HTTPS 与限流——模型端点直接暴露等于把你的 API Key 和数据敞开给全网。
22.5 运行时能力全景
AgentOS 的设计是"按需加能力"，每个能力一个开关：
| 需求 | 配置 |
|---|---|
| 存储/追踪有默认数据库 | db=... |
| JWT 授权 | authorization=True |
| 存储执行追踪 | tracing=True（下一章展开） |
| 把实例暴露为 MCP Server | mcp_server=True |
| 定时任务 | scheduler=True |
| 挂进已有 FastAPI 应用 | base_app=existing_app |
base_app= 是渐进式迁移的关键：已有 FastAPI 项目的业务路由保持不变，只把 /agents/* 等前缀挂载进来，避免为了引入 Agno 重写整个服务。
22.6 本章小结
- AgentOS = 官方运行时，把组件包成你自托管的 FastAPI 应用，
get_app()+serve()两步起服务； - 执行端点统一为
POST /{component}/{id}/runs，靠session_id维持会话线程； - 先查
GET /info（含auth_mode）再调用，是客户端的标准握手动作； - 生产必开
authorization=True；tracing、mcp_server、scheduler、base_app都是单开关能力。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. 调用 AgentOS 中某个 Agent 的正确 REST 端点是？
2. 如何让两次 HTTP 调用属于同一个会话线程？
3. 客户端判断 AgentOS 实例使用哪种鉴权方式，应该调用？
4. 已有大型 FastAPI 项目想引入 AgentOS 而不重写路由，最合适的做法是？
🛠️ 动手实践
- 用
uvicorn agent_os:app --port 7777以外的方式（如 gunicorn 多 worker）启动本章的服务，并用两个不同session_id各发三条消息验证会话隔离。 - 编写一个 Python 客户端函数
chat(message, session_id)，封装对/agents/support-agent/runs的调用并处理 SSE 流式响应。 - 开启
authorization=True后分别带和不带凭证调用/info与/agents/*/runs，记录哪些端点公开、哪些被拦截。
服务上线后如何看清它内部发生了什么？下一章：可观测性与调试。