4000行代码撑起一个Agent框架？nanobot架构深度解析

Chat Platforms (13 channels)
         ↓
   MessageBus (asyncio.Queue)
    inbound / outbound
         ↓
   AgentLoop（核心控制面）
    ├── SessionManager  → 历史对话加载
    ├── ContextBuilder  → system prompt + messages 组装
    └── while loop: LLM → tool calls → execute → append → LLM...
         ↓
  ┌──────────────────────────────┐
  │ ToolRegistry   SubagentMgr   │
  │ exec/fs/web    asyncio.Task  │
  │ spawn/mcp/cron               │
  └──────────────────────────────┘

async def _run_agent_loop(self, initial_messages, on_progress):
    messages = initial_messages
    while iteration < self.max_iterations:   # 默认 40 次
        response = await self.provider.chat(
            messages=messages,
            tools=self.tools.get_definitions(),
            model=self.model,
        )
        if response.has_tool_calls:
            for tool_call in response.tool_calls:
                result = await self.tools.execute(tool_call.name, tool_call.arguments)
                messages = append_tool_result(messages, tool_call.id, result)
        else:
            final_content = response.content
            break
    return final_content

if result.startswith("Error"):
    return result + "\n\n[Analyze the error above and try a different approach.]"

class Tool(ABC):
    @property
    @abstractmethod
    def name(self) -> str: ...
    
    @property
    @abstractmethod
    def description(self) -> str: ...
    
    @property
    @abstractmethod
    def parameters(self) -> dict: ...  # JSON Schema
    
    @abstractmethod
    async def execute(self, **kwargs) -> str: ...  # 硬约束：必须返回 str

class MyDatabaseTool(Tool):
    @property
    def name(self): return "query_database"
    
    @property
    def description(self): return "执行只读 SQL 查询"
    
    @property
    def parameters(self):
        return {
            "type": "object",
            "properties": {"sql": {"type": "string"}},
            "required": ["sql"]
        }
    
    async def execute(self, sql: str, **kwargs) -> str:
        rows = await db.fetch(sql)
        return "\n".join(str(r) for r in rows)

self.tools.register(MyDatabaseTool(connection_string="..."))

---
name: weather
description: Get current weather and forecasts (no API key required).
metadata: {"nanobot":{"emoji":"🌤️","requires":{"bins":["curl"]}}}
---

# Weather

## wttr.in (primary)

```bash
curl -s "wttr.in/London?format=3"

系统通过 `shutil.which` 检查 `bins` 里的工具是否存在。如果存在，这个 skill 的 `available` 标记为 `true`。

### Progressive Loading：用文件系统做懒加载

关键机制：系统不把所有 skill 内容塞进 system prompt，而是注入一个 XML 索引：

```xml
<skills>
  <skill available="true">
    <name>weather</name>
    <description>Get current weather...</description>
    <location>/path/to/skills/weather/SKILL.md</location>
  </skill>
  <skill available="false">
    <name>github</name>
    <description>Interact with GitHub via gh CLI</description>
    <requires>Binary: gh</requires>
    <location>/path/to/skills/github/SKILL.md</location>
  </skill>
</skills>

async def consolidate(self, session, provider, model):
    response = await provider.chat(
        messages=[...],
        tools=_SAVE_MEMORY_TOOL,   # save_memory(history_entry, memory_update)
        model=model,
    )
    # LLM 决定写什么进 MEMORY.md 和 HISTORY.md

async def spawn(self, task, label, origin_channel, origin_chat_id):
    task_id = str(uuid.uuid4())[:8]
    asyncio.create_task(
        self._run_subagent(task_id, task, label, origin)
    )
    return f"Subagent [{label}] started (id: {task_id}). I'll notify you when done."

async def _announce_result(self, task_id, label, task, result, origin):
    msg = InboundMessage(
        channel="system",
        chat_id=f"{origin['channel']}:{origin['chat_id']}",
        content=f"[Subagent '{label}' completed]\nResult: {result}",
    )
    await self.bus.publish_inbound(msg)

class MCPToolWrapper(Tool):
    def __init__(self, session, server_name, tool_def):
        self._name = f"mcp_{server_name}_{tool_def.name}"  # 命名空间隔离
        self._description = tool_def.description
        self._parameters = tool_def.inputSchema
    
    async def execute(self, **kwargs) -> str:
        result = await asyncio.wait_for(
            self._session.call_tool(self._original_name, arguments=kwargs),
            timeout=self._tool_timeout,
        )
        return "\n".join(block.text for block in result.content)

{
  "tools": {
    "mcp_servers": {
      "my_server": {
        "command": "npx",
        "args": ["-y", "@my/mcp-server"],
        "tool_timeout": 30
      }
    }
  }
}