agno AgentOS 怎么提交后台 run 轮询状态、取消任务并在断线后用 SSE 续接【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno当你通过 HTTP 调用 agno AgentOS 服务时会碰到三类实际问题长任务不想让请求一直挂着希望提交后就拿到一个 run 标识再去轮询任务跑偏了需要主动取消并确认它确实停了流式客户端中途断线希望能从断点把 SSE 事件补回来。agno 的 cookbook 中cookbook/05_agent_os/04_run_lifecycle/目录提供了这三个能力的完整可运行示例background_run.py后台提交 轮询、cancel_run.py取消 确认终态、sse_reconnect.py断线后用/resume续接。本文按这三条路径逐一演示提交、轮询、取消与续接的完整 HTTP 契约。前提条件来自 README已按仓库scripts/demo_setup.sh初始化 cookbook 环境并设置export OPENAI_API_KEYyour-keyyour-key替换为你自己的 OpenAI 密钥示例代码用OpenAIResponses(idgpt-5.5)作为模型后台 run 依赖数据库持久化 run 状态被服务的 Agent 上必须配置数据库示例用SqliteDb否则轮询和取消读不到持久化状态。后台模式不支持 remote agents每个示例文件都是一个 FastAPI 服务都监听http://localhost:7777同一时间只能启动一个。启动服务端一个文件既是服务又是客户端三个示例结构一致不带参数运行时启动 AgentOS 服务带--demo参数时切换为调用已运行服务的客户端。以 background_run.py 为例终端 1 启动服务端.venvs/demo/bin/python cookbook/05_agent_os/04_run_lifecycle/background_run.py服务端的 Agent 定义保留了后台 run 的关键前提——挂在Agent上的db参数db SqliteDb( idbackground-run-db, db_filetmp/agent_os_background_run.db, ) background_run_agent Agent( idbackground-run-agent, nameBackground Run Agent, modelOpenAIResponses(idgpt-5.5), dbdb, instructionsAnswer clearly and keep the final response under three paragraphs., ) agent_os AgentOS(idbackground-run-os, agents[background_run_agent]) app agent_os.get_app()agent_os.serve(appapp)默认在localhost:7777上监听serve的host/port默认值为localhost和7777也可被环境变量AGENT_OS_HOST、AGENT_OS_PORT覆盖。提交后台 run 并轮询到终态终端 2 运行客户演示.venvs/demo/bin/python cookbook/05_agent_os/04_run_lifecycle/background_run.py --demo契约分三步示例代码httpx.AsyncClientbase_urlhttp://localhost:7777逐一对应1. 提交。POST /agents/{agent_id}/runs表单字段带backgroundtrue、streamfalse、session_idresponse await client.post( f/agents/{AGENT_ID}/runs, data{ message: Explain why persisted background runs are useful., background: true, stream: false, session_id: SESSION_ID, }, )成功判定HTTP 状态码必须是202响应体status必须是PENDING示例代码对这两个条件不满足就直接抛RuntimeError。从响应体里取出run_id和session_id二者是后续轮询、取消、续接的全部凭据。2. 轮询嵌套 run 路由。GET /agents/{agent_id}/runs/{run_id}用查询参数带session_id循环读取status字段直到进入终态集合{CANCELLED, COMPLETED, ERROR}示例每 0.5 秒轮询一次总超时 120 秒response await client.get( f/agents/{AGENT_ID}/runs/{run_id}, params{session_id: session_id}, ) run response.json() status run[status] if status in {CANCELLED, COMPLETED, ERROR}: break await asyncio.sleep(0.5)注意 README 强调的一个细节RunStatus有七个成员PENDING、RUNNING、PAUSED、COMPLETED、CANCELLED、ERROR、REGENERATEDstatus 要从持久化的 run output 里读run 事件流中除了WorkflowPaused之外没有事件携带status字段——所以轮询嵌套路由是正确姿势而不是解析事件。3. 判定完成。示例最后断言completed[status] COMPLETED并打印completed.get(content)作为 run 的结果。RunStatus的完整定义见 run/base.py其中REGENERATED是/continue?regeneratetrue留下的标记状态。取消一个已接受的后台 runcancel_run.py 的前半段与上面完全相同同样是backgroundtrue、streamfalse提交同样等待 HTTP 202 /PENDING。差别在提交之后、轮询之前多一步取消调用.venvs/demo/bin/python cookbook/05_agent_os/04_run_lifecycle/cancel_run.py .venvs/demo/bin/python cookbook/05_agent_os/04_run_lifecycle/cancel_run.py --democancel_response await client.post( f/agents/{AGENT_ID}/runs/{run_id}/cancel, params{session_id: session_id}, ) cancel_response.raise_for_status()即POST /agents/{agent_id}/runs/{run_id}/cancel用查询参数带session_id。随后沿用同一个轮询函数读持久化状态直到观察到CANCELLED示例以cancelled[status] ! CANCELLED抛错作为失败判定。两条需要写进客户端逻辑的边界取消不等于丢弃已完成的输出。README 说明被取消的 run 会保留已做完的工作agents 和 teams 保留在content里workflow 保留在step_results里取消后流式端点的收尾事件是 completed不是 cancelled。取消会发出成对事件先 cancelled、再 completed所以如果你用 SSE 观察取消流的结束信号是 completed 事件真正可靠的终态仍以轮询到的持久化CANCELLED为准。断线后用 SSE 续接跟踪 event_index再打 /resume第三条路径用 sse_reconnect.py。它有一个重要前提AgentOSClient目前没有暴露 resume 方法所以示例刻意用裸httpx处理原始 SSE 响应和续接请求。.venvs/demo/bin/python cookbook/05_agent_os/04_run_lifecycle/sse_reconnect.py .venvs/demo/bin/python cookbook/05_agent_os/04_run_lifecycle/sse_reconnect.py --demo run .venvs/demo/bin/python cookbook/05_agent_os/04_run_lifecycle/sse_reconnect.py --demo continue--demo run演示新后台流的续接--demo continue演示「先暂停、后台继续、再断线续接」的更复杂路径。两者的共同机制如下。1. 消费流时持续记录坐标。提交时用backgroundtrue、streamtrue打开 SSE 流客户端逐事件记录三类状态事件里的run_id、session_id以及最大的event_index。示例刻意只读前 2 个事件就断开EVENTS_BEFORE_DISCONNECT 2模拟网络断开state[run_id] event.get(run_id) or state[run_id] state[session_id] event.get(session_id) or state[session_id] if event.get(event_index) is not None: state[last_event_index] event[event_index]SSE 解析本身是标准的event:/data:行协议[DONE]作为结束标记不解析为事件。2. 重连。用run_id、session_id和断点前收到的最后event_index调POST /agents/{agent_id}/runs/{run_id}/resumedata {session_id: session_id} if last_event_index is not None: data[last_event_index] str(last_event_index) async with client.stream( POST, f/agents/{AGENT_ID}/runs/{run_id}/resume, datadata, ) as response: response.raise_for_status() async for event in iter_sse(response): ...resume 流会补齐断线期间错过的剩余事件直到 run 结束。验证方式示例给得很直接resume 必须返回至少一个事件否则抛The resume endpoint returned no events然后打印本次收到的event_index范围Resumed event_index range: min..max你可以用它核对断点前后的事件没有丢、没有乱。3.continue路径的额外一步。--demo continue场景里 Agent 带一个requires_confirmationTrue的工具初次运行会在RunPaused事件处停住。RunPaused事件携带run_id、session_id和待确认的tools数组客户端把其中requires_confirmation的条目加上confirmedTrue后原样连同新决定一起提交到嵌套/continue路由agent 的字段名是tools并带backgroundtrue、streamtrue之后断线再 resume 的流程与上面一致。README 提醒continue 后 run 仍可能再次暂停应基于 status 循环处理不要认为一次/continue就是终态且同一个 run 在/continue后run_id和session_id都保持不变。验证清单与限制汇总三条路径的成功判据都来自示例代码的断言可直接作为你自己客户端的验收条件路径关键请求成功判据后台提交 轮询POST /agents/{id}/runsbackgroundtrue→GET /agents/{id}/runs/{run_id}?session_id...提交返回 HTTP 202 且statusPENDING轮询到COMPLETED结果在 run output 的content取消POST /agents/{id}/runs/{run_id}/cancel?session_id...取消调用raise_for_status()通过轮询到CANCELLED已完成的输出仍保留在contentSSE 续接记录event_index→POST /agents/{id}/runs/{run_id}/resume带session_id与last_event_indexresume 流返回非空事件打印的event_index范围覆盖断点之后限制方面README 明确了两条硬性约束后台 run 要求被服务的 Agent 配置数据库因为 detached task 和 poll 都读持久化状态且该模式不支持 remote agents。端口方面三个示例都固定监听 7777启动多个示例前必须先停掉上一个。如果你想接着做暂停/继续human-in-the-loop场景README 指向了同目录体系下的 user_input.py 演示 team 的 pause/continue而 checkpointtool-batch断点续跑、后台 hooks 等相邻能力则分别在同目录的checkpoints.py和hooks_in_background.py中不在本文范围内。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考