自托管 Agent Protocol 服务器：基于 Deep Agents 构建异步子 Agent 服务端与 Supervisor 完整指南

自托管 Agent Protocol 服务器：基于 Deep Agents 构建异步子 Agent 服务端与 Supervisor 完整指南

【免费下载链接】deepagents The batteries-included agent harness. 项目地址: https://gitcode.com/GitHub_Trending/de/deepagents

异步子 Agent（async subagent）模式让主 Agent 可以像分发后台任务一样，把研究工作委托给运行在远端服务器上的另一个 Agent：任务一经提交立即返回 task_id，主 Agent 随后按需轮询状态、追加指令或取消任务，全程无需阻塞等待。本文以 Deep Agents 仓库中的 async-subagent-server 示例 为核心，完整讲解"自托管 Agent Protocol 服务器 + Supervisor 接线"这一模式的落地方法：你将掌握服务端端点的实现原理、Supervisor 侧的配置方式、底层 AsyncSubAgentMiddleware 的五个工具契约，以及如何把示例中的研究 Agent 替换为你自己的 Agent。

模式全景：Async Subagent 的两端

Deep Agents 的异步子 Agent 模式天然包含两个对端，示例目录 恰好把两端都做了最小可运行的实现：

• server.py — 一个基于 FastAPI 的自托管 Agent Protocol 服务器，把 Deep Agents 研究 Agent 暴露为异步子 Agent；
• supervisor.py — 一个交互式 REPL，演示如何通过 LangGraph SDK 连接到该服务器并驱动后台研究任务。

两端通过 Agent Protocol 的 HTTP 契约通信。Supervisor 侧提交任务后立即拿到 task_id，任务在服务端后台执行，Supervisor 可随时检查状态、发送新指令或取消任务。从 middleware/async_subagents.py 的模块注释可以看到，异步子 Agent 通过 LangGraph SDK 在远端 Agent Protocol 服务器上启动后台 run，与阻塞等待的同步子 Agent 形成鲜明对比。

示例项目本身是一个独立的 uv 工程，pyproject.toml 声明了运行所需的依赖：deepagents（通过 [tool.uv.sources] 以 editable 方式指向仓库内 libs/deepagents）、fastapi、uvicorn、langchain-anthropic、langgraph、langgraph-sdk、httpx 与 python-dotenv。

前置条件与环境准备

开始前需要准备两个环境变量（详见 README）：

代码层面，服务端在启动时会通过 load_dotenv(Path(__file__).parent / ".env")（server.py）从示例目录下的 .env 文件加载环境变量，Supervisor 也有同样的加载逻辑（supervisor.py）。README 建议 cp .env.example .env 后填写密钥；由于代码本身对缺失的 TAVILY_API_KEY 有降级处理，即使只配置 ANTHROPIC_API_KEY，整个示例也能跑通——此时 lifespan 启动钩子 会打印一条 [warn] TAVILY_API_KEY not set — using stub search 的提示。

快速开始：四步跑通完整链路

按 README 的 Quickstart 章节（README）逐步执行：

1. 安装依赖：

cd examples/async-subagent-server
uv sync

2. 配置环境变量：

cp .env.example .env
# 填入 ANTHROPIC_API_KEY（可选填 TAVILY_API_KEY）

3. 启动服务端：

uv run uvicorn server:app –port 2024

4. 另开一个终端启动 Supervisor：

cd examples/async-subagent-server
ANTHROPIC_API_KEY=… uv run python supervisor.py

Supervisor 启动后会打印 Supervisor connected to researcher at http://localhost:2024，随后进入 > 提示符的 REPL（supervisor.py）。README 给出了五类可直接体验的指令：

> research the latest developments in quantum computing
> check status of <task-id>
> update <task-id> to focus on commercial applications only
> cancel <task-id>
> list all tasks

这五类指令恰好一一对应 Deep Agents 异步子 Agent 中间件提供的五个工具：start_async_task、check_async_task、update_async_task、cancel_async_task 与 list_async_tasks。

Agent Protocol 端点对照：协议层契约

README 的 Implemented endpoints 表格列出了服务端必须实现的 Agent Protocol 端点，这是 Deep Agents 异步子 Agent 中间件（经由 LangGraph SDK）实际会调用的接口：

每个端点都能在 server.py 中找到一一对应的实现，且都标注了中间件侧调用它的时机：

• GET /ok（L215-L218）：返回 {"ok": true}，用于连通性探活。
• POST /threads（L221-L231）：中间件的 start_async_task 在创建 run 之前调用，生成 UUID 作为 thread_id 并写入 threads 表，返回空消息列表。
• POST /threads/{thread_id}/runs（L234-L296）：start_async_task（新任务）与 update_async_task（带新指令重新运行）都会调用它。请求体可携带 multitask_strategy 与 input.messages，服务端解析出最后一条 user 消息追加进线程历史，创建 run 后通过 asyncio.ensure_future 以 fire-and-forget 方式执行 Agent，立即返回 status: "pending"。
• GET /threads/{thread_id}/runs/{run_id}（L299-L305）：check_async_task 轮询任务是否完成，返回 run 的实时状态。
• GET /threads/{thread_id}（L308-L317）：run 进入 success 后，check_async_task 通过它读取线程状态；LangGraph SDK 会从 values['messages'] 中提取最终结果，因此服务端在 run 执行器 中把 assistant 回复同时写入 messages 与 values_ 两个字段。
• POST /threads/{thread_id}/runs/{run_id}/cancel（L320-L332）：cancel_async_task 调用，把 run 标记为 cancelled。

服务端实现深度解析：状态、Agent 与执行生命周期

零配置的内存 SQLite 存储

服务端使用进程内共享的 SQLite（:memory:，check_same_thread=False）做持久化，"无需文件、无需初始化"（server.py）。启动时由 lifespan 钩子调用 _init_db() 自动建表（L52-L78）：

• threads 表：thread_id、created_at、messages（{role, content} 对象的 JSON 数组）、values_（以 JSON 存储线程最终状态，即 values.messages）；
• runs 表：run_id、thread_id、assistant_id、status（pending | running | success | error | cancelled）、created_at、error。

内置研究 Agent 与 web_search 工具

服务端通过 create_deep_agent 构造研究 Agent（L153-L164）：

_agent = create_deep_agent(
model=ChatAnthropic(model="claude-sonnet-4-5"),
system_prompt=(
"You are a thorough research agent. Investigate topics using web search and produce "
"a well-structured research summary (300–500 words). Cite sources where possible.\\n\\n"
"If you receive new instructions mid-conversation, follow them immediately without "
"asking for clarification — discard prior work and start fresh on the new task."
),
tools=[web_search],
)

系统提示词特意强调"收到新指令立即执行、丢弃旧工作"，这与 update_async_task 的"中断重启"语义相呼应。web_search 工具（L119-L150）做了双分支处理：检测到 TAVILY_API_KEY 时通过 httpx 异步调用 Tavily 搜索 API（max_results=5）；未配置时返回格式化的 stub 结果，保证示例离线可跑。

Run 的执行与状态机

_execute_run（L169-L197）是后台执行的核心：先把 run 置为 running，然后调用 _agent.ainvoke({"messages": [HumanMessage(user_message)]})；成功后取出最后一条消息作为 assistant 回复，追加进线程消息历史并同步写 values_，最后把 run 置为 success；任何异常都会把 run 置为 error 并记录错误信息。客户端全程靠轮询 GET /threads/{thread_id}/runs/{run_id} 感知状态迁移。

create_run 还实现了 multitask_strategy: "interrupt" 语义（L250-L259）：当请求体携带该策略时，先把同线程下所有 running 的 run 置为 cancelled、清空线程 values_，再启动新 run——这正是 Supervisor 侧 update_async_task 发送"重定向研究"指令时的实际行为。

取消的边界语义

需要注意 cancel 端点 只把 run 在数据库中标记为 cancelled，并不会真正中断正在执行的 Agent 调用；源码注释明确指出，如需真正的取消要自行接入 asyncio.Task 取消机制。对演示场景而言，状态层面的一致已经足够。

Supervisor 侧接线：AsyncSubAgent 配置与系统提示词

Supervisor 的核心是把远端服务声明为一个 AsyncSubAgent 规格，然后通过 create_deep_agent(subagents=…) 挂载（supervisor.py）：

RESEARCHER_URL = os.environ.get("RESEARCHER_URL", "http://localhost:2024")

async_subagents: list[AsyncSubAgent] = [
{
"name": "researcher",
"description": (
"A research agent that investigates any topic using web search. "
"Runs in the background and returns a detailed summary."
),
"graph_id": "researcher",
"url": RESEARCHER_URL,
"headers": {"x-auth-scheme": "custom"},
},
]

supervisor = create_deep_agent(
model=ChatAnthropic(model="claude-sonnet-4-5"),
checkpointer=MemorySaver(),
system_prompt=(…),
subagents=async_subagents,
)

AsyncSubAgent 规格的字段定义在 middleware/async_subagents.py：

• name：异步子 Agent 的唯一标识，同时是工具调用时的 subagent_type；
• description：主 Agent 依据它决定何时委托任务；
• graph_id：远端服务器上对应的 graph 名称或 assistant ID；
• url（可选）：Agent Protocol 服务器的地址，缺省时使用 LangGraph SDK 默认端点，也可以省略以启用本地 ASGI 传输（仅限 ainvoke 异步入口）；
• headers（可选）：访问远端服务器的附加请求头。注意中间件的 _resolve_headers（L186-L196）默认会补上 x-auth-scheme: langsmith，本示例显式覆盖为 custom，展示自托管场景下如何按需覆盖。

此外，graph.py 会把传入 subagents= 的规格按类型拆分为内联子 Agent 与异步子 Agent 两组，后者最终组装进 AsyncSubAgentMiddleware（graph.py）。checkpointer=MemorySaver() 为 Supervisor 提供了会话内线程状态保存能力，配合 thread_id 保证多轮对话共享同一会话（supervisor.py）。

Supervisor 的系统提示词（supervisor.py）把五个工具的使用协议写成了显式规则：仅当用户出现 research、investigate、look into、find out 等触发词才启动研究员；启动后立即回报 task_id 并停止，不得立刻查状态；状态更新永远调用工具而非凭记忆汇报；不循环轮询；始终展示完整 task_id。这套提示词是演示 REPL 稳定工作的关键。

换入你自己的 Agent

README 的 "Swap in your own agent" 章节（README）指出：Agent Protocol 协议层与具体 Agent 实现解耦，只需替换 server.py 中的 create_deep_agent 调用即可：

_agent = create_deep_agent(
model=ChatAnthropic(model="claude-sonnet-4-5"),
system_prompt="You are a …",
tools=[your_tool],
)

服务端代码对 Agent 的唯一约束是：接受一个 messages 数组，并返回一个包含 messages 数组的对象（server.py 的注释）。这意味着你可以在不触碰任何 HTTP 端点的情况下，把研究 Agent 换成代码审查、文档写作或任意自定义工具的 Agent，Supervisor 侧只需保持 graph_id、name 与服务端一致。

中间件原理：五个工具与任务状态机

异步子 Agent 的能力由 AsyncSubAgentMiddleware 提供（middleware/async_subagents.py），它默认注入五个工具（见 _build_async_subagent_tools，L815-L837）：

任务记录以 AsyncTask 形式持久化在 Agent 状态 async_tasks 字段中（通过 reducer 合并，L122-L129），因此能够跨上下文压缩/卸载存活。中间件的单元测试位于 tests/unit_tests/test_async_subagents.py，其中验证了"至少一个子 Agent 的初始化约束"与"恰好创建五个工具"等契约。

端到端验证：无真实 LLM 的 HTTP 契约测试

示例自带 test_server.py，通过 FastAPI TestClient 在不打真实 LLM 的前提下验证 Agent Protocol 的 HTTP 契约——把 server._agent.ainvoke patch 成返回固定响应的 AsyncMock。覆盖场景包括：

• 健康检查与线程创建；
• 创建 run 后状态为 pending；
• 完整生命周期：建线程 → 建 run → 等后台任务结束 → 轮询得到 success → 读取线程 values.messages 拿到 assistant 回复；
• 取消 run 后状态变为 cancelled；
• multitask_strategy="interrupt" 会把同线程正在运行的第一个 run 取消；
• 对不存在的线程与 run 返回 404。

其中"完整生命周期"测试（test_server.py）精确对应了 README 端点表格中 GET /threads/{thread_id} 的用途说明："成功时读取 values.messages"。这套测试可以当作你自己实现 Agent Protocol 服务端时的契约参考。

生产化注意事项

README 末尾的免责声明（README）明确：本示例仅用于演示自托管异步子 Agent 模式，不包含认证、限流等生产所需能力。落地到生产环境时需要自行补齐：

• 服务端接入身份认证（中间件侧可通过 AsyncSubAgent.headers 携带令牌）；
• 增加限流、超时与重试策略；
• 将内存 SQLite 替换为持久化存储，并考虑多实例部署下的并发一致性；
• 为 cancel 接入真正的 asyncio.Task 取消，而不是仅更新状态标记。

延伸阅读

如果你想继续深入这套模式，仓库内还有以下相关资源：异步子 Agent 中间件的完整实现与文档字符串（middleware/async_subagents.py）、create_deep_agent 对 subagents= 参数的统一处理逻辑（graph.py），以及 LangChain 官方的 LangChain Academy 免费课程可作为学习 Agent 生态的补充资料。

【免费下载链接】deepagents The batteries-included agent harness. 项目地址: https://gitcode.com/GitHub_Trending/de/deepagents

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考