Agno AgentOS 上手实战：用 HTTP 一次打通 Agent、Team、Workflow 与知识库的完整服务端

Agno AgentOS 上手实战用 HTTP 一次打通 Agent、Team、Workflow 与知识库的完整服务端【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno 的AgentOSagno.os是一个把 Agent、Team、Workflow、知识库、会话存储统一挂载进单一 FastAPI 服务的运行时载体。本篇指南基于仓库中的 01 Getting Started 课程带你用两个脚本完整体验 AgentOS先用full_os.py在 7777 端口启动一个全量 OS再用run_over_http.py通过原始 HTTP 调用完成服务发现、非流式运行、SSE 流式运行与会话列表查询。读完你就能掌握 AgentOS 的端点地图、表单式运行协议、SSE 事件格式与分页会话返回结构并可以直接把它作为后续搭建 Agent 平台的第一块脚手架。课程概览AgentOS 到底serve了什么本课程要回答一个核心问题AgentOS 被 serve 之后实际暴露了哪些能力。答案在 01_getting_started/README.md 中非常明确——它把四类运行时原语primitive挂载到一起原语说明Agent单智能体绑定知识库支持检索增强回答Team多智能体协作团队协调成员后返回单一结论Workflow把 Agent 包装成可复用的工作流步骤Knowledge本地知识库SQLite 存内容元数据 Chroma 存向量随后通过两个文件展开完整闭环文件职责full_os.py在端口 7777 上 serve 一个 Agent、一个 Team、一个 Workflow 和一个 Chroma 支撑的知识库run_over_http.py调用/config做服务发现执行非流式与 SSE 流式 Agent 运行并列出持久化后的会话课程的验证记录见 TEST_LOG.md实测中/health返回200 ok/config报告 OS IDgetting-started-os、数据库getting-started-db、agent/team/workflow 三个组件齐全/knowledge/config显示 ChromaDB 后端并支持 vector、keyword、hybrid 三种检索。端点地图一个 OS 暴露的完整 HTTP 表面文档给出了一张清晰的端点地图映射了每类原语被挂载后引入的路由原语引入的路由AgentGET /agents、POST /agents/getting-started-agent/runsTeamGET /teams、POST /teams/getting-started-team/runsWorkflowGET /workflows、POST /workflows/getting-started-workflow/runsKnowledge/knowledge/content、/knowledge/search、/knowledge/configPersistenceGET /sessionsDiscoveryGET /config两点需要特别记住的协议细节Agent 运行入参是表单字段form fields不是 JSON body——对应源码中message: str Form(...)、stream: bool Form(True)、session_id、user_id均为Form类型参数见 agents/router.py。会话列表是分页返回的结构统一为{ data: [...], meta: {...} }——meta中包含page、limit、total_count、total_pages见 session/session.py。从源码结构看app.py 的_add_built_in_routesAgentOS 还会自动挂载更多内置路由/health健康检查、/info轻量 OS 元数据无需认证、/docs与/openapi.json交互式文档可通过settings.docs_enabled开关。此外还有/sessions/{session_id}、/sessions/{session_id}/runs等会话详情路由以及团队与工作流对应的.../runs/{run_id}/cancel、.../runs/{run_id}/resume、.../runs/{run_id}/continue等运行控制端点。前置条件零外部服务的本地运行按文档要求本课程只需要一个环境变量与两个本地组件OPENAI_API_KEYAgent 与 Team 都使用OpenAIResponses(idgpt-5.5)知识库向量化使用OpenAIEmbedder(idtext-embedding-3-small)因此需要有效的 OpenAI Key。SQLite会话与知识内容持久化本地文件无需外部数据库服务。Chroma向量数据库同样本地运行无需外部服务。此外建议使用仓库演示环境demo课程脚本用.venvs/demo/bin/python运行说明 demo venv 中已安装 agno、fastapi、uvicorn、httpx、chromadb 等依赖。第一步编写并启动全量 AgentOSfull_os.py是本课程的服务端其构建流程分为四段每一段都对应 AgentOS 的一类核心抽象。1. 数据库与知识库from agno.agent import Agent from agno.db.sqlite import SqliteDb from agno.knowledge import Knowledge from agno.knowledge.embedder.openai import OpenAIEmbedder from agno.models.openai import OpenAIResponses from agno.os import AgentOS from agno.team import Team from agno.vectordb.chroma import ChromaDb from agno.workflow.step import Step from agno.workflow.workflow import Workflow db SqliteDb( idgetting-started-db, db_filetmp/getting_started.db, ) knowledge Knowledge( nameGetting Started Knowledge, descriptionDocuments uploaded during the getting-started lesson., contents_dbdb, vector_dbChromaDb( pathtmp/getting_started_chroma, collectiongetting_started, embedderOpenAIEmbedder(idtext-embedding-3-small), ), )要点说明SqliteDb是会话表agno_sessions与知识内容表的持久化载体db_file指向本地文件运行后可在tmp/下看到生成的 SQLite 文件。Knowledge采用双库架构contents_dbSQLite保存文档内容与元数据vector_dbChroma保存向量ChromaDb的collection与path定义了向量集合名与存储目录。OpenAIEmbedder(idtext-embedding-3-small)负责把文档切块向量化。2. Agent、Team、Workflowassistant Agent( idgetting-started-agent, nameGetting Started Agent, modelOpenAIResponses(idgpt-5.5), knowledgeknowledge, search_knowledgeTrue, instructionsAnswer clearly and use the knowledge base when it is relevant., markdownTrue, ) assistant_team Team( idgetting-started-team, nameGetting Started Team, modelOpenAIResponses(idgpt-5.5), members[assistant], instructionsCoordinate the available specialist and return one concise answer., markdownTrue, ) answer_workflow Workflow( idgetting-started-workflow, nameGetting Started Workflow, descriptionRun the assistant as a reusable workflow step., steps[Step(nameAnswer Question, agentassistant)], )要点说明search_knowledgeTrue开启知识库检索使 Agent 能基于Knowledge内容回答。Team的members[assistant]把一个 Agent 作为团队成员挂进团队运行时会由协调模型统一调度。Workflow的steps[Step(nameAnswer Question, agentassistant)]把同一个 Agent 包装为可复用的工作流步骤实现一次定义、多处编排。三者都显式指定了id这些 id 将直接出现在 URL 路径中如/agents/getting-started-agent/runs也是会话查询时component_id的取值。3. 组装 AgentOS 并启动agent_os AgentOS( idgetting-started-os, descriptionOne AgentOS exposing every core runtime primitive., dbdb, agents[assistant], teams[assistant_team], workflows[answer_workflow], knowledge[knowledge], ) app agent_os.get_app() if __name__ __main__: agent_os.serve(appfull_os:app, reloadTrue)底层原理AgentOS.__init__会完成组件初始化——为未显式指定 db 的 Agent/Team/Workflow 注入 OS 级默认db、检查重复 ID、把组件注册进 registry见 app.py。get_app()则构建 FastAPI 应用并组装生命周期数据库初始化、httpx 客户端清理、可选 MCP/调度器等见 app.py。serve()的默认参数为hostlocalhost、port7777并支持AGENT_OS_HOST/AGENT_OS_PORT环境变量覆盖见 app.pyreloadTrue启用 uvicorn 热重载方便开发期修改即生效。课程给出的启动命令.venvs/demo/bin/python cookbook/05_agent_os/01_getting_started/full_os.py启动后可先做两个快速验证文档也建议用 curl 直接查看发现文档curl http://localhost:7777/config curl http://localhost:7777/healthGET /config的响应由 router.py 构建包含os_id、description、available_models、databases、session、memory、knowledge、evals、metrics、agents、teams、workflows、interfaces等字段——这就是 AgentOS 的完整发现文档。第二步用纯 HTTP 走通 Agent 全流程run_over_http.py是客户端脚本通过httpx以原始 HTTP 方式与运行中的服务交互全程不依赖 agno 的 Python 客户端。它依次完成四件事服务发现、非流式运行、SSE 流式运行、会话列表查询。先看它定义的关键常量BASE_URL http://localhost:7777 AGENT_ID getting-started-agent SESSION_ID getting-started-http-session USER_ID getting-started-userSESSION_ID与USER_ID是刻意固定的两次运行共用同一会话与用户从而在最后一步能查询到同一个会话被两次运行复用的持久化证据。服务发现GET /configdef show_config(client: httpx.Client) - None: response client.get(/config) response.raise_for_status() config response.json() print(fAgentOS: {config[os_id]}) print(fAgents: {[agent[id] for agent in config[agents]]}) print(fTeams: {[team[id] for team in config[teams]]}) print(fWorkflows: {[workflow[id] for workflow in config[workflows]]})/config返回的agents、teams、workflows是组件摘要列表客户端按id字段提取即可得到全部已注册组件。这一步的价值在于动态发现客户端不必硬编码组件清单只需解析发现文档就能知道 OS 暴露了哪些可运行对象。非流式运行表单 POST 完整 JSON 响应def run_non_streaming(client: httpx.Client) - None: response client.post( f/agents/{AGENT_ID}/runs, data{ message: What does this AgentOS expose? Answer in one sentence., stream: false, session_id: SESSION_ID, user_id: USER_ID, }, ) response.raise_for_status() result response.json() print(fRun ID: {result[run_id]}) print(fSession ID: {result[session_id]}) print(fResponse: {result[content]})这里验证了文档强调的协议要点请求体是application/x-www-form-urlencoded字段包括message必填、streamfalse关闭流式、session_id、user_id。响应为完整 JSON可直接取出run_id、session_id、content。对照源码agents/router.pyPOST /agents/{agent_id}/runs还支持更多可选表单字段files多模态上传支持图片/音频/视频/PDF/CSV/DOCX 等、files_metadataJSON 数组与 files 按位置对应、version组件版本、background后台运行需数据库、factory_input动态构建 Agent 的工厂参数。stream字段默认值为True因此非流式请求必须显式传streamfalse。SSE 流式运行逐步消费事件def run_streaming(client: httpx.Client) - None: with client.stream( POST, f/agents/{AGENT_ID}/runs, data{ message: Give me a short streamed welcome to AgentOS., stream: true, session_id: SESSION_ID, user_id: USER_ID, }, ) as response: response.raise_for_status() if text/event-stream not in response.headers.get(content-type, ): raise RuntimeError(Expected an SSE response) event_name message for line in response.iter_lines(): if line.startswith(event: ): event_name line[7:] elif line.startswith(data: ): payload json.loads(line[6:]) if event_name RunContent and payload.get(content): print(payload[content], end, flushTrue) elif event_name RunCompleted: print(f\nCompleted run: {payload[run_id]})SSE 流式协议要点Content-Type 必须是text/event-stream脚本对此做了显式校验。事件格式为标准 SSEevent: 事件名与data: JSON交替出现。脚本识别两种关键事件RunContent携带增量content逐字打印与RunCompleted携带run_id标志运行结束。源码中的 OpenAPI 示例还展示了RunStarted事件见 agents/router.py即完整事件序列大致为RunStarted → RunContent(多次) → RunCompleted。从源码结构看AgentOS 的 SSE 输出由format_sse_event等工具函数生成见 os/utils.py 中导入列表流式与非流式走同一运行内核只是响应封装方式不同。会话列表分页查询持久化结果def show_sessions(client: httpx.Client) - None: response client.get( /sessions, params{ type: agent, component_id: AGENT_ID, user_id: USER_ID, }, ) response.raise_for_status() payload response.json() print(fSessions found: {payload[meta][total_count]}) for session in payload[data]: print(f- {session[session_id]})GET /sessions支持多维过滤与分页见 session/session.py查询参数说明默认值type会话类型agent、team、workflow不传返回全部无component_id按组件 IDagent/team/workflow ID过滤无user_id按用户 ID 过滤无session_name按名称模糊匹配无limit每页条数20page页码从 1 开始1sort_by排序字段created_atsort_orderasc/descdescdb_id/table指定数据库或数据表无返回结构为PaginatedResponsedata是会话对象数组含session_id、session_name、session_state、created_at、updated_at等字段meta包含page、limit、total_count、total_pages。由于两次运行复用了SESSION_IDgetting-started-http-session此处应恰好查询到这一个持久化会话——这正是会话被真正写入了 SQLite的直接证据。一键走完整个闭环客户端主流程run_over_http.pyif __name__ __main__: with httpx.Client(base_urlBASE_URL, timeout120.0) as http_client: show_config(http_client) print(\nNon-streaming run) run_non_streaming(http_client) print(\nStreaming run) run_streaming(http_client) print(\nStored sessions) show_sessions(http_client)注意timeout120.0模型推理可能耗时较长客户端设置了 120 秒超时以避免流式中途断开。运行命令.venvs/demo/bin/python cookbook/05_agent_os/01_getting_started/run_over_http.py预期输出依次为发现文档中三个组件 ID → 非流式运行的单句回答含 run_id/session_id→ 流式逐字输出的欢迎语与完成事件 → 查询到的getting-started-http-session会话。深入源码AgentOS 的服务端装配逻辑为了让挂载即服务的理解更加扎实可以再看两个关键的底层机制。内置路由的装配AgentOS._add_built_in_routesapp.py展示了每个 OS 实例自动获得的路由集合Core/home、/health、/info、/config、WebSocket/workflows/ws组件agent、team、workflow 三类 router数据面session、media、memory、learnings、evals、metrics、knowledge、traces、database 等 router。而这些 router 只有在传入了相应依赖时才启用例如db为None时/components、/schedules、/approvals、/service-accounts会返回503提示传入 db 以启用而不是 404——这种设计让缺失功能的诊断信息更加明确见 app.py。OS 级默认值与组件注入AgentOS.__init__的_initialize_agents/_initialize_teams/_initialize_workflowsapp.py做三件重要的事把 OS 级db注入到所有未自行指定 db 的组件上保证会话持久化默认可用把store_events True写到每个 Agent/Team/Workflow这是内置路由能查询会话与运行记录的前提递归收集团队与工作流中的 MCP 工具统一纳入生命周期管理。这也是为什么本课程中full_os.py只需给AgentOS传一次dbdbAgent/Team/Workflow 就能共享同一 SQLite 会话库。常见问题与排查思路streamfalse被忽略总是返回流式stream表单字段默认是True见 agents/router.py非流式请求务必显式传stream: false。SSE 响应解析为空先检查Content-Type是否为text/event-stream事件按行读取时event:与data:行的解析顺序很重要当前事件名会保持到下一个event:出现为止。查询不到会话确认两次运行使用了相同的session_id与user_id且/sessions的type、component_id与组件实际类型、ID 一致。端口冲突serve()默认localhost:7777可通过AGENT_OS_HOST、AGENT_OS_PORT环境变量覆盖或直接修改serve()的host/port参数见 app.py。想看完整 API 面启动后访问http://localhost:7777/docsSwagger UI或http://localhost:7777/openapi.json可浏览全部端点与参数说明。小结通过full_os.py与run_over_http.py两个文件你已经完成了一次对 Agno AgentOS 的最小闭环体验构建Agent/Team/Workflow/Knowledge→ 挂载AgentOS→ 发现/config→ 运行非流式 SSE→ 持久化/sessions。这套模式可以直接推广到团队路由POST /teams/{team_id}/runs与工作流路由POST /workflows/{workflow_id}/runs——它们的表单协议与会话模型与 Agent 完全同构。进一步可以探索 05_agent_os 目录下的其他课程数据库、Python 客户端、运行生命周期、MCP 等把 AgentOS 从单机演示演进为带认证、队列、调度能力的完整 Agent 平台。延伸阅读服务端完整示例full_os.pyHTTP 客户端完整示例run_over_http.py实测验证记录TEST_LOG.mdAgentOS 核心实现libs/agno/agno/os/app.py发现文档与核心路由libs/agno/agno/os/router.pyAgent 运行与列表路由libs/agno/agno/os/routers/agents/router.py会话分页查询实现libs/agno/agno/os/routers/session/session.py知识库路由libs/agno/agno/os/routers/knowledge/knowledge.py【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考