第 6 章 · AgentOS 服务化平台

前 5 章讲的是「一个 agent/team/workflow 在进程里怎么跑」。这章讲 怎么把它们变成生产服务:AgentOS 接收一组 agent,生成一个完整的 FastAPI 应用——带 REST 端点、流式、权限、调度。这是 Agno 区别于纯「agent 库」的地方。

6.1 它解决的问题

你本地写好了 agent,现在要让前端/其他服务通过 HTTP 调用它:列出会话、发起一次 run、流式拿输出、暂停时人审批、定时跑任务、按用户隔离权限……这些「服务层」的活,AgentOS 全包了。

6.2 入口:AgentOS → get_app()

用法极简(回顾 index 的例子):你把 agent 列表交给 AgentOS,调 get_app() 拿到一个标准 FastAPI app。

# 示意,非源码
from agno.os import AgentOS
os_app = AgentOS(agents=[my_agent], teams=[my_team], workflows=[my_wf])
app = os_app.get_app()       # 标准 FastAPI app
# 之后用 uvicorn 跑,或 os_app.serve(...) 一步到位

AgentOS.__init__(os/app.py:222)把三类资源存下来(os/app.py:303-305):

# os/app.py:303 —— AgentOS 持有的三类资源
self.agents = agents        # List[Agent | RemoteAgent | AgentProtocol | AgentFactory]
self.teams = teams
self.workflows = workflows

注意类型里有 Remote* 和 *Factory ——意味着资源可以是远程的(跑在别处的 agent)或工厂(运行时再造),不只本地实例。

6.3 路由:50+ 端点是怎么挂上的

_add_built_in_routes(os/app.py:509)把一组 router 逐个加进 app。每个 router 管一类资源:

# os/app.py:515 —— 内建路由逐个挂载(节选)
self._add_router(app, get_health_router(health_endpoint="/health"))
self._add_router(app, get_info_router(self))
self._add_router(app, get_base_router(self, settings=self.settings))
self._add_router(app, get_agent_router(self, settings=self.settings, registry=self.registry))
self._add_router(app, get_team_router(self, settings=self.settings, registry=self.registry))
self._add_router(app, get_workflow_router(self, settings=self.settings))

各 router 的工厂分散在 os/router.py 和 os/routers/(如 approvals 路由 get_approval_router,在 os/app.py:47 import、:473 使用)。每个 router 暴露「列出/获取/运行/管理」这一类资源的端点,合起来就是文档说的 50+ 端点。

结构图(请求怎么找到你的 agent):

  HTTP 请求 (POST /agents/{id}/runs)
        │
        ▼
  FastAPI app  ← AgentOS.get_app()
        │ 路由分发
        ▼
  agent_router  (os/router.py / os/routers/)
        │ 鉴权(RBAC scope)→ 找到对应 Agent
        ▼
  agent.arun(input, ...)   ← 第 1 章那条流水线
        │
        ▼
  SSE 流 / JSON 响应 ──▶ 客户端

6.4 流式:SSE 与 WebSocket

第 1 章的 streaming 路径(_arun_stream 沿途 yield RunOutputEvent)在这里派上用场:agent router 把这些事件包成 SSE 推给客户端。Workflow 还有专门的 WebSocket 端点(get_websocket_router → workflow_websocket_endpoint,os/router.py:278、:297)。

6.5 RBAC:基于 scope 的权限

AgentOS 内建权限模型,基于 scope 字符串(os/scopes.py)。scope 有三种形态(os/scopes.py:5-7):

形态	例子	含义
全局资源 scope	agents:read	对某类资源的某动作
单资源 scope	agents:<id>:run	对特定资源的某动作
特殊 scope	admin	管理员/通配

AgentOSScope 枚举(os/scopes.py:36)列出所有合法 scope,ParsedScope(:89)把一个 scope 字符串解析成 (type, resource, resource_id, action),用 is_global_resource_scope / is_per_resource_scope(:100、:105)判断它管哪一层。鉴权基于 JWT(os/auth.py),从 token 里取 scope 比对端点要求。

这套设计支持文档说的「多用户、多租户隔离」——同一个 OS 上不同用户凭 token 里的 scope 只能碰自己被授权的资源。

6.6 生命周期与调度

FastAPI 的 lifespan 被用来管理「启动/关停时该做的事」,AgentOS 把多个 lifespan 合并(_combine_app_lifespans,os/app.py:174):

• DB lifespan(db_lifespan,os/app.py:127)—— 连接/关闭数据库。
• MCP lifespan(mcp_lifespan,:85)—— 管理 MCP 工具连接的生命周期。
• HTTP client lifespan(:97)、scheduler lifespan(scheduler_lifespan,:141)。
• 关停时还会 _drain_cancel_persist_tasks(:106)把未完成的取消/持久化任务排干(默认 30s 超时)。

调度(cron): scheduler_lifespan 启动内建调度器(scheduler/ 目录),实现文档说的「cron 调度 + 后台任务,无需外部基础设施」。

6.7 人审端点

第 2 章的 HITL 在这里有对应的 HTTP 面:get_approval_router(os/app.py:47 import、:473 重建)提供审批端点——当一次 run 因 requires_confirmation 暂停,客户端通过审批端点确认/拒绝,触发 continue_run 续跑。

6.8 巧妙之处

• 本地 agent 零改动变服务。 你写 agent 时根本不碰 HTTP;AgentOS 在外面包一层就上线了——「先在 notebook 里跑通,再 AgentOS(agents=[...]).get_app()」。
• router 可重建(resync)。 resync / _reprovision_routers(os/app.py:423、:449)能在运行时重新挂载路由——支持热更新资源集合。
• 功能未装就给「禁用」路由。 _get_disabled_feature_router(os/app.py:198)在某特性缺依赖时挂一个返回明确错误的占位路由,而不是直接 404 让人困惑。
• lifespan 合并 + 排干。 多个子系统(DB/MCP/HTTP/scheduler)的启停被合并管理,关停时主动排干在途任务(:106),减少「关服务时丢数据」。

6.9 边界

• AgentOS 是 FastAPI;要嵌进已有 FastAPI app,可传 base_app(os/app.py:244、:313),路由会加到你的 app 上。
• scope 体系兼容旧的 system:* 别名(os/scopes.py:21 注释),迁移时要注意。

6.10 代码地图

主题	文件	符号
OS 入口	libs/agno/agno/os/app.py	AgentOS、AgentOS.get_app、AgentOS.serve
内建路由挂载	libs/agno/agno/os/app.py	_add_built_in_routes、_add_router
路由工厂	libs/agno/agno/os/router.py	get_base_router、get_info_router、get_websocket_router
资源路由	libs/agno/agno/os/routers/	get_agent_router、get_approval_router(等)
RBAC	libs/agno/agno/os/scopes.py	AgentOSScope、ParsedScope
鉴权	libs/agno/agno/os/auth.py	(JWT 鉴权)
生命周期	libs/agno/agno/os/app.py	db_lifespan、mcp_lifespan、scheduler_lifespan、_combine_app_lifespans
调度	libs/agno/agno/scheduler/	(cron 调度器)