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

01

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               │  └──────────────────────────────┘

02

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.]"

03

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="..."))

04

---name: weatherdescription: Get current weather andforecasts (no API key required).metadata: {"nanobot":{"emoji":"🌤️","requires":{"bins":["curl"]}}}---
# Weather
## wttr.in (primary)
```bashcurl -s "wttr.in/London?format=3"

系统通过 `shutil.which` 检查 `bins` 里的工具是否存在。如果存在，这个 skill 的 `available` 标记为 `true`。
### Progressive Loading：用文件系统做懒加载
关键机制：系统不把所有 skill 内容塞进 system prompt，而是注入一个 XML 索引：
```xml<skills>  <skillavailable="true">    <name>weather</name>    <description>Get current weather...</description>    <location>/path/to/skills/weather/SKILL.md</location>  </skill>  <skillavailable="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>

05

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

06

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)

07

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      }    }  }}

08

09