nanobot 源码解读 · 08 | 多 Agent 委派：一个 Agent 如何把活派给子 Agent

async def spawn(self, task, label=None, origin_channel="cli", session_key=None,
                workspace_scope=None, *, runtime=None) -> str:
    """后台派活，立即返回，结果稍后通过消息总线通知。"""
    task_id = str(uuid.uuid4())[:8]
    ...
    bg_task = asyncio.create_task(
        self._run_subagent(task_id, task, display_label, origin, status, runtime, ...)
    )
    self._running_tasks[task_id] = bg_task
    ...
    return f"Subagent [{display_label}] started (id: {task_id}). I'll notify you when it completes."

async def run_inline(self, task, ..., *, runtime=None) -> str:
    """同步派活，await 拿到结果再返回（直接作为工具结果回给主 Agent）。"""
    inline_task = asyncio.create_task(self._run_subagent(...))
    try:
        result = await inline_task
        if status.phase == "error" or status.stop_reason in {"error", "tool_error"}:
            return ToolResult.error(result)   # 失败 → 包装成工具错误
        return result
    finally:
        ...清理...

async def _run_subagent(self, task_id, task, label, origin, status, runtime,
                        origin_message_id, workspace_scope, *, announce=True) -> str:
    try:
        root = workspace_scope.project_path if workspace_scope is not None else self.workspace
        cfg = None
        if workspace_scope is not None:
            cfg = self._subagent_tools_config()
            cfg.restrict_to_workspace = workspace_scope.restrict_to_workspace
        tools = self._build_tools(tools_config=cfg)          # ① 独立工具注册表
        system_prompt = self._build_subagent_prompt(workspace=root)  # ② 子 Agent 专属系统提示
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user",   "content": task},             # ③ 任务就是第一条用户消息
        ]
        # ④ 权限红线照旧：用 contextvars 注入（呼应 02 篇）
        request_token = bind_request_context(RequestContext(
            channel=origin["channel"], chat_id=origin["chat_id"],
            message_id=origin_message_id, session_key=..., runtime=runtime))
        token = bind_workspace_scope(workspace_scope) if workspace_scope is not None else None
        try:
            result = await self.runner.run(AgentRunSpec(    # ⑤ 复用 02 篇引擎！
                initial_messages=messages,
                tools=tools,
                runtime=runtime,
                max_iterations=self.max_iterations,
                max_tool_result_chars=self.max_tool_result_chars,
                hook=_SubagentHook(task_id, status),
                checkpoint_callback=_on_checkpoint,
                session_key=sess_key,
                workspace=root,
                llm_timeout_s=llm_timeout,
            ))
        finally:
            if token is not None: reset_workspace_scope(token)
            reset_request_context(request_token)
        # ⑥ 根据 stop_reason 决定最终回传文本 + 是否 announce
        ...
        if announce:
            await self._announce_result(task_id, label, task, final_result, origin, final_status, ...)
        return final_result
    except Exception as e:
        ...

def _build_tools(self, workspace=None, tools_config=None) -> ToolRegistry:
    root = self.workspace if workspace is None else workspace
    registry = ToolRegistry()
    cfg = tools_config if tools_config is not None else self._subagent_tools_config()
    ctx = ToolContext(
        config=cfg,
        workspace=str(root.resolve()),
        exec_session_manager=self._exec_session_manager,
        file_state_store=FileStates(),                       # 独立的文件状态
        workspace_sandbox=workspace_sandbox_status(
            restrict_to_workspace=cfg.restrict_to_workspace, workspace=root),
    )
    ToolLoader().load(ctx, registry, scope="subagent")      # 仅加载 subagent 作用域的工具
    return registry

def _subagent_tools_config(self) -> ToolsConfig:
    return ToolsConfig(
        exec=self.tools_config.exec,
        web=self.tools_config.web,
        file=self.tools_config.file,
        restrict_to_workspace=self.restrict_to_workspace,    # 默认限制在工作区
    )

async def _announce_result(self, task_id, label, task, result, origin, status, origin_message_id):
    status_text = "completed successfully" if status == "ok" else "failed"
    announce_content = render_template(
        "agent/subagent_announce.md",
        label=label, status_text=status_text, task=task, result=result,
    )
    # 以 system 消息注入，触发主 Agent 的「中途注入」机制
    override = origin.get("session_key") or f"{origin['channel']}:{origin['chat_id']}"
    msg = InboundMessage(
        channel="system",
        sender_id="subagent",
        chat_id=f"{origin['channel']}:{origin['chat_id']}",
        content=announce_content,
        session_key_override=override,                    # 对齐主 Agent 的有效 session
        metadata={"injected_event": "subagent_result", "subagent_task_id": task_id},
    )
    await self.bus.publish_inbound(msg)                  # 丢回主 Agent 的消息总线