Parent agent Subagent +——————+ +——————+ | messages=[…] | | messages=[] | <– fresh | | dispatch | | | tool: task | ———-> | while tool_use: | | prompt=”…” | | call tools | | | summary | append results | | result = “…” | <———- | return last text | +——————+ +——————+

主Agent接收任务

class SubagentManager:

self._running_tasks: dict[str, asyncio.Task | None] = {} self._session_tasks: dict[str, set[str]] = {}

async def spawn( 
self, task: str, # 子代理要执行的任务描述 label: str | None = None, # 显示标签（用于用户识别） origin_channel: str = "cli", # 原始渠道（用于结果通知） origin_chat_id: str = "direct", # 原始聊天 ID session_key: str | None = None, # 会话键（用于会话级取消） 
 
) -> str:

task_id = str(uuid.uuid4())[:8] # 生成 8 字符的 UUID4，如 “a1b2c3d4” display_label = label or task[:30] + (“…” if len(task) > 30 else “”)

origin = {“channel”: origin_channel, “chat_id”: origin_chat_id}

bg_task = asyncio.create_task( 
self._run_subagent(task_id, task, display_label, origin) 
 
) self._running_tasks[task_id] = bg_task

if session_key: 
self._session_tasks.setdefault(session_key, set()).add(task_id)

def _cleanup(_: asyncio.Task) -> None: self._running_tasks.pop(task_id, None) if session_key and (ids := self._session_tasks.get(session_key)): ids.discard(task_id) if not ids: del self._session_tasks[session_key] 

bg_task.add_done_callback(_cleanup)

return f”Subagent [{display_label}] started (id: {task_id}). I’ll notify you when it completes.”

tools = ToolRegistry() allowed_dir = self.workspace if self.restrict_to_workspace else None 
tools.register(ReadFileTool(workspace=self.workspace, allowed_dir=allowed_dir)) tools.register(WriteFileTool(workspace=self.workspace, allowed_dir=allowed_dir)) tools.register(EditFileTool(workspace=self.workspace, allowed_dir=allowed_dir)) tools.register(ListDirTool(workspace=self.workspace, allowed_dir=allowed_dir)) tools.register(ExecTool(
 
working_dir=str(self.workspace), timeout=self.exec_config.timeout, restrict_to_workspace=self.restrict_to_workspace, path_append=self.exec_config.path_append, 
 
)) tools.register(WebSearchTool(api_key=self.brave_api_key)) tools.register(WebFetchTool())

system_prompt = self._build_subagent_prompt(task) messages = [ 
{"role": "system", "content": system_prompt}, {"role": "user", "content": task}, 
 
]

# Subagent 
Current Time
 
{now} ({tz})
 
You are a subagent spawned by main agent to complete a specific task.
 
Rules
 
      
     
        
 
         
Stay focused - complete only the assigned task, nothing else
 
         
Your final response will be reported back to main agent
 
         
Do not initiate conversations or take on side tasks
 
         
Be concise but informative in your findings
 
        
 
What You Can Do
 
      
     
        
 
         
Read and write files in workspace
 
         
Execute shell commands
 
         
Search web and fetch web pages
 
         
Complete task thoroughly
 
        
 
What You Cannot Do
 
      
     
        
 
         
Send messages directly to users (no message tool available)
 
         
Spawn other subagents
 
         
Access main agent’s conversation history
 
        
 
Workspace
 
Your workspace is at: {workspace} Skills are available at: {workspace}/skills/ (read SKILL.md files as needed)
 
When you have completed the task, provide a clear summary of your findings or actions.

max_iterations = 15 # 子代理的最大迭代次数（主 Agent 是 40） iteration = 0 final_result: str | None = None 
while iteration < max_iterations:
 
iteration += 1 response = await self.provider.chat( messages=messages, tools=tools.get_definitions(), model=self.model, temperature=self.temperature, max_tokens=self.max_tokens, )

if response.has_tool_calls: # 构建工具调用消息 tool_call_dicts = [ { "id": tc.id, "type": "function", "function": { "name": tc.name, "arguments": json.dumps(tc.arguments, ensure_ascii=False), }, } for tc in response.tool_calls ] messages.append({ "role": "assistant", "content": response.content or "", "tool_calls": tool_call_dicts, }) # 执行工具 for tool_call in response.tool_calls: result = await tools.execute(tool_call.name, tool_call.arguments) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "name": tool_call.name, "content": result, })

else: final_result = response.content break

if final_result is None: final_result = "Task completed but no final response was generated."

logger.info("Subagent [{}] completed successfully", task_id) 

await self._announce_result(task_id, label, task, final_result, origin, “ok”)

except Exception as e: 
error_msg = f"Error: {str(e)}" logger.error("Subagent [{}] failed: {}", task_id, e) await self._announce_result(task_id, label, task, error_msg, origin, "error")

async def _announce_result( self, task_id: str, label: str, task: str, result: str, origin: dict[str, str], status: str, 

) -> None:

status_text = “completed successfully” if status == “ok” else “failed” 
announce_content = f”“”[Subagent ‘{label}’ {status_text}] Task: {task} Result: {result}
 
Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not mention technical details like “subagent” or task IDs.“””

msg = InboundMessage( 
channel="system", # 使用 system 渠道标识 sender_id="subagent", # 标识来自子代理 chat_id=f"{origin['channel']}:{origin['chat_id']}", # 原始渠道和聊天 ID content=announce_content, # 通知内容 
 
)
 
await self.bus.publish_inbound(msg)

async def cancel_by_session(self, session_key: str) -> int: 
"""Cancel all subagents for a given session. Returns count cancelled.""" tasks = [ self._running_tasks[tid] for tid in self._session_tasks.get(session_key, []) if tid in self._running_tasks and not self._running_tasks[tid].done() ] for t in tasks: t.cancel() if tasks: await asyncio.gather(*tasks, return_exceptions=True) return len(tasks)

def get_running_count(self) -> int: """Return number of currently running subagents.""" return len(self._running_tasks)

# SubagentManager 初始化时 

self.provider = provider # 和 Main Agent 使用同一个实例

Subagent 执行时

response = await self.provider.chat(

messages=messages, tools=tools.get_definitions(), model=self.model, temperature=self.temperature, max_tokens=self.max_tokens, 

)

self.bus = bus # 和 Main Agent 使用同一个实例 
Subagent 完成任务后
 
await self.bus.publish_inbound(msg) # 通知 Main Agent

self.workspace = workspace # 和 Main Agent 使用同一个路径 
Subagent 中的工具
 
tools.register(ReadFileTool(workspace=self.workspace, …))

tools = ToolRegistry() tools.register(ReadFileTool(…)) # 文件读取 tools.register(WriteFileTool(…)) # 文件写入 tools.register(EditFileTool(…)) # 文件编辑 tools.register(ListDirTool(…)) # 目录列表 tools.register(ExecTool(…)) # Shell 执行 tools.register(WebSearchTool(…)) # 网络搜索 tools.register(WebFetchTool(…)) # 网页获取

# 除了上述 7 个工具外，还包含： 
          
     
            
 
             
MessageTool (发送消息给渠道)
 
             
SpawnTool (派生子代理)
 
             
CronTool (引用 cron_service)
 
             
MCPServersTool (连接 MCP 服务器)
 
             
SkillsTool (动态加载技能)
 
            

def _build_subagent_prompt(self, task: str) -> str: return f”“”# Subagent 
Current Time
 
{now} ({tz})
 
You are a subagent spawned by main agent to complete a specific task.
 
Rules
 
          
     
            
 
             
Stay focused - complete only assigned task, nothing else
 
             
Your final response will be reported back to main agent
 
             
Do not initiate conversations or take on side tasks
 
             
Be concise but informative in your findings
 
            
 
What You Can Do
 
          
     
            
 
             
Read and write files in workspace
 
             
Execute shell commands
 
             
Search web and fetch web pages
 
             
Complete task thoroughly
 
            
 
What You Cannot Do
 
          
     
            
 
             
Send messages directly to users (no message tool available)
 
             
Spawn other subagents
 
             
Access main agent’s conversation history …“””
 
            

# nanobot 🐈 
You are nanobot, a helpful AI assistant.
 
Runtime
 
{system} {machine}, Python {version}
 
Workspace
 
Your workspace is at: {workspace_path}
 
nanobot Guidelines
 
          
     
            
 
             
State intent before tool calls…
 
             
Before modifying a file, read it first… …
 
            
 
(还会包含 Bootstrap Files、Long-term Memory、Skills 等)

messages = [ 
{"role": "system", "content": subagent_system_prompt}, {"role": "user", "content": task}, # 只有当前任务描述 
 
]

messages = [ 
{"role": "system", "content": main_system_prompt}, *history, # 完整的会话历史（可能数百条） {"role": "user", "content": runtime_context}, # 运行时元数据 {"role": "user", "content": current_message}, 
 
]

# Subagent 不直接发送结果给用户 final_result = response.content await self._announce_result(task_id, label, task, final_result, origin, “ok”) 
通过 MessageBus 通知 Main Agent，Main Agent 再转发给用户

# Main Agent 直接发送结果给用户 
await self.bus.publish_outbound(OutboundMessage(
 
channel=msg.channel, chat_id=msg.chat_id, content=final_content, 
 
))

agent = AgentLoop( 
bus=bus, # 共享 provider=provider, # 共享 workspace=workspace, # 共享 model=config.agents.defaults.model, temperature=config.agents.defaults.temperature, max_tokens=config.agents.defaults.max_tokens, max_iterations=40, # 完整对话需要更多迭代 memory_window=config.agents.defaults.memory_window, session_manager=SessionManager(...), # 独有：会话持久化 cron_service=cron, # 独有：定时任务服务 mcp_servers=config.tools.mcp_servers, # 独有：MCP 服务器 channels_config=config.channels, # 独有：渠道配置 
 
)

# 通过 _run_subagent() 内部直接创建，不通过构造函数 async def _run_subagent(self, task_id, task, label, origin): 
# 工具集：受限的 7 个工具 tools = ToolRegistry() tools.register(ReadFileTool(...)) # ... (不含 MessageTool, SpawnTool, CronTool) # System Prompt：聚焦式提示 system_prompt = self._build_subagent_prompt(task) # 消息：只有当前任务，无历史 messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": task}, ] # 迭代次数：15 次 max_iterations = 15 # 循环执行 while iteration < max_iterations: response = await self.provider.chat(...) # 使用共享的 provider # ...