Nanobot —— 工具系统与 MCP 集成
#Nanobot, Agent

本文档拆解 Nanobot 的工具系统设计，涵盖 ToolRegistry 动态注册机制、内置工具分类、MCP 集成原理与自定义工具开发。

1. 工具系统概述

Nanobot 的工具系统基于注册表模式，所有工具统一通过 ToolRegistry 管理，Agent Loop 只通过注册表调用工具，不感知具体实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
┌──────────────────────────────────────────────────────┐
│                   ToolRegistry                       │
│                                                      │
│  ┌─────────────┐  ┌──────────────┐  ┌────────────┐  │
│  │ FileSystem  │  │    Shell     │  │    Web     │  │
│  │ Tools       │  │    Tool      │  │   Tools    │  │
│  └─────────────┘  └──────────────┘  └────────────┘  │
│  ┌─────────────┐  ┌──────────────┐  ┌────────────┐  │
│  │  Search     │  │    MCP       │  │  Custom    │  │
│  │  Tools      │  │   Tools      │  │  Tools     │  │
│  └─────────────┘  └──────────────┘  └────────────┘  │
│                                                      │
│  list_schemas() → LLM 工具定义列表                   │
│  execute(name, input) → 工具结果                     │
└──────────────────────────────────────────────────────┘

2. ToolRegistry 设计

2.1 核心接口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class ToolRegistry:
    _tools: dict[str, BaseTool] = {}

    def register(self, tool: BaseTool) -> None:
        """注册工具"""
        self._tools[tool.name] = tool

    def get(self, name: str) -> BaseTool:
        """获取工具实例"""
        if name not in self._tools:
            raise ToolNotFoundError(f"Tool '{name}' not registered")
        return self._tools[name]

    def list_schemas(self) -> list[dict]:
        """生成所有工具的 JSON Schema，供 LLM 使用"""
        return [tool.to_schema() for tool in self._tools.values()]

    async def execute(self, name: str, input: dict) -> str:
        """执行工具，返回结果字符串"""
        tool = self.get(name)
        try:
            return await tool.run(**input)
        except Exception as e:
            return f"Tool error: {type(e).__name__}: {str(e)}"

2.2 BaseTool 基类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class BaseTool(ABC):
    name: str           # 工具唯一名称（snake_case）
    description: str    # 工具描述（决定 LLM 何时调用）
    parameters: dict    # JSON Schema 参数定义

    @abstractmethod
    async def run(self, **kwargs) -> str:
        """工具执行逻辑，返回字符串结果"""

    def to_schema(self) -> dict:
        """生成 LLM 工具定义格式"""
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": {
                "type": "object",
                "properties": self.parameters,
                "required": [...],
            }
        }

2.3 并发工具执行

当 LLM 在一次响应中返回多个工具调用时，ToolRegistry 并发执行：

1
2
3
4
5
6
7
8
9
async def execute_parallel(
    self, tool_calls: list[ToolCallRequest]
) -> list[str]:
    """并发执行多个工具调用"""
    tasks = [
        self.execute(call.name, call.input)
        for call in tool_calls
    ]
    return await asyncio.gather(*tasks, return_exceptions=True)

3. 内置工具分类

3.1 文件系统工具（filesystem.py）

工具名 功能 主要参数
read_file 读取文件内容（含 PDF/Office） path, encoding
write_file 创建或覆写文件 path, content
edit_file 精确字符串替换（safer） path, old_string, new_string
list_dir 列出目录内容 path, recursive
delete_file 删除文件（需确认） path 
1
2
3
4
5
6
7
8
9
10
11
class ReadFileTool(BaseTool):
    name = "read_file"
    description = "读取指定路径的文件内容。支持文本文件、PDF 和 Office 文档。"

    async def run(self, path: str, encoding: str = "utf-8") -> str:
        p = Path(path)
        if p.suffix == ".pdf":
            return self._read_pdf(p)
        elif p.suffix in (".docx", ".xlsx", ".pptx"):
            return self._read_office(p)
        return p.read_text(encoding=encoding)

3.2 Shell 工具（shell.py）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class ExecTool(BaseTool):
    name = "exec"
    description = "执行 Shell 命令并返回输出。"

    async def run(self, command: str, timeout: int = 30) -> str:
        proc = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(
            proc.communicate(), timeout=timeout
        )
        return stdout.decode() or stderr.decode()

3.3 搜索工具（search.py）

工具名 功能 等同命令
glob 文件路径模式匹配 find . -name "*.py"
grep 文件内容搜索（正则） grep -r "pattern" . 
1
2
3
4
5
6
7
8
9
class GlobTool(BaseTool):
    name = "glob"
    description = "用 glob 模式匹配文件路径，按修改时间排序返回。"

    async def run(self, pattern: str, path: str = ".") -> str:
        import glob
        files = glob.glob(f"{path}/{pattern}", recursive=True)
        files.sort(key=lambda f: os.path.getmtime(f), reverse=True)
        return "\n".join(files[:100])

3.4 Web 工具（web.py）

工具名 功能 说明
web_search DuckDuckGo 搜索 返回摘要和链接
web_fetch 获取网页内容 返回纯文本（Markdown 格式）

3.5 交互工具（ask.py）

1
2
3
4
5
6
7
8
class AskUserTool(BaseTool):
    name = "ask_user"
    description = "向用户提问，等待用户回复后继续。适用于需要用户确认的操作。"

    async def run(self, question: str) -> str:
        # 暂停 Agent，等待用户通过渠道回复
        response = await self.interaction_manager.ask(question)
        return response

3.6 定时任务工具（cron.py）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
class CronTool(BaseTool):
    name = "schedule_task"
    description = "创建定时任务，在指定时间或按 cron 表达式重复执行。"

    async def run(
        self,
        task: str,
        schedule: str,  # cron 表达式或自然语言（"每天9点"）
        channel: str,
    ) -> str:
        cron_expr = self.parse_schedule(schedule)
        self.cron_service.add(
            task=task,
            schedule=cron_expr,
            channel=channel,
        )
        return f"任务已创建：{task}，调度：{cron_expr}"

4. MCP 集成

4.1 MCP 工具包装原理

Nanobot 将 MCP Server 暴露的工具自动包装为 BaseTool，注册到 ToolRegistry，Agent 无需区分本地工具和 MCP 工具。

1
2
3
4
5
6
7
8
9
10
11
MCP Server（独立进程）
    │  stdio / HTTP+SSE
    ▼
MCPClient（nanobot/agent/tools/mcp.py）
    │  list_tools() → 工具清单
    │  call_tool(name, args) → 执行结果
    ▼
MCPToolWrapper（将每个 MCP 工具包装为 BaseTool）
    │
    ▼
ToolRegistry.register(wrapped_tool)

4.2 MCPToolWrapper 实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class MCPToolWrapper(BaseTool):
    """将单个 MCP 工具包装为 BaseTool"""

    def __init__(self, mcp_client: MCPClient, tool_def: dict):
        self.name = tool_def["name"]
        self.description = tool_def["description"]
        self.parameters = tool_def["inputSchema"]["properties"]
        self._client = mcp_client

    async def run(self, **kwargs) -> str:
        result = await self._client.call_tool(self.name, kwargs)
        # MCP 返回 TextContent / ImageContent 等
        return "\n".join(
            item.text for item in result.content
            if hasattr(item, "text")
        )

4.3 MCP Server 配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// config.json 中配置 MCP Server
{
  "tools": {
    "mcp_servers": [
      {
        "name": "filesystem",
        "command": "npx",
        "args": [
          "-y",
          "@modelcontextprotocol/server-filesystem",
          "/allowed/path"
        ],
        "transport": "stdio"
      },
      {
        "name": "github",
        "command": "npx",
        "args": ["-y", "@modelcontextprotocol/server-github"],
        "env": { "GITHUB_TOKEN": "${GITHUB_TOKEN}" },
        "transport": "stdio"
      },
      {
        "name": "remote-tool",
        "url": "http://localhost:3000/mcp",
        "transport": "http_sse"
      }
    ]
  }
}

4.4 MCP 工具发现流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
Nanobot 启动
    │
    ▼
MCPManager.initialize()
    │
    ├─ 启动所有配置的 MCP Server 进程
    │
    ├─ 调用 list_tools() 获取每个 Server 的工具清单
    │
    ├─ 为每个工具创建 MCPToolWrapper
    │
    └─ 注册到 ToolRegistry（自动可用）
    │
    ▼
Agent 启动，所有 MCP 工具可用

5. 自定义工具开发

5.1 开发步骤

Step 1: 继承 BaseTool

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from nanobot.agent.tools.base import BaseTool

class DatabaseQueryTool(BaseTool):
    name = "query_database"
    description = (
        "查询业务数据库。当需要获取订单、用户、产品等业务数据时使用。"
        "支持 SQL 查询，返回 JSON 格式结果。"
    )
    parameters = {
        "sql": {
            "type": "string",
            "description": "SQL 查询语句（只支持 SELECT）"
        },
        "limit": {
            "type": "integer",
            "description": "最大返回行数",
            "default": 100
        }
    }

    def __init__(self, db_url: str):
        self.db_url = db_url

    async def run(self, sql: str, limit: int = 100) -> str:
        # 安全校验：只允许 SELECT
        if not sql.strip().upper().startswith("SELECT"):
            return "Error: 只允许 SELECT 查询"

        async with aiopg.connect(self.db_url) as conn:
            cursor = await conn.cursor()
            await cursor.execute(f"{sql} LIMIT {limit}")
            rows = await cursor.fetchall()
            return json.dumps(rows, ensure_ascii=False)

Step 2: 注册到 ToolRegistry

1
2
3
4
5
# 在 Nanobot 初始化时注册
from nanobot.agent.tools.registry import tool_registry

db_tool = DatabaseQueryTool(db_url="postgresql://...")
tool_registry.register(db_tool)

5.2 工具设计最佳实践

原则 说明 示例
description 清晰 说明何时调用、参数含义 “当用户询问订单信息时使用”
职责单一 一个工具做一件事 查询和修改分开
错误友好 返回可读错误而非抛异常 return f"Error: {e}"
参数精简 只暴露必要参数 内部默认值不作为参数
幂等查询 只读工具无副作用 SELECT 只查不改

5.3 工具注册检查

1
2
3
4
5
6
7
8
9
10
11
12
13
# 开发时验证工具是否正确注册和可用
async def test_tool():
    # 检查 schema 生成是否正确
    schemas = tool_registry.list_schemas()
    db_schema = next(s for s in schemas if s["name"] == "query_database")
    print(json.dumps(db_schema, indent=2))

    # 测试执行
    result = await tool_registry.execute(
        "query_database",
        {"sql": "SELECT COUNT(*) FROM orders"}
    )
    print(result)

6. 工具执行安全

6.1 危险工具保护

对于可能造成不可逆影响的工具（删除文件、执行 Shell），通过 Hook 系统添加确认拦截：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class SafetyHook(AgentHook):
    DANGEROUS_TOOLS = {"delete_file", "exec"}
    DANGEROUS_PATTERNS = {"rm -rf", "sudo", "DROP TABLE"}

    async def before_execute_tools(self, tool_calls):
        for call in tool_calls:
            if call.name in self.DANGEROUS_TOOLS:
                # 暂停，等待用户确认
                confirmed = await self.ask_user(
                    f"⚠️ 即将执行: {call.name}({call.input})，确认继续？"
                )
                if not confirmed:
                    raise ToolExecutionDenied(f"用户拒绝执行 {call.name}")

            if call.name == "exec":
                cmd = call.input.get("command", "")
                for pattern in self.DANGEROUS_PATTERNS:
                    if pattern in cmd:
                        raise ToolExecutionDenied(f"命令包含危险模式: {pattern}")

6.2 沙箱隔离（sandbox.py）

可选启用沙箱，将 Shell 命令执行隔离在 Docker 容器中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class SandboxedExecTool(ExecTool):
    """在 Docker 容器中执行命令"""

    async def run(self, command: str, timeout: int = 30) -> str:
        result = await asyncio.create_subprocess_exec(
            "docker", "run", "--rm",
            "--memory=256m",           # 内存限制
            "--cpus=0.5",              # CPU 限制
            "--network=none",          # 禁用网络
            "nanobot-sandbox:latest",  # 沙箱镜像
            "bash", "-c", command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(
            result.communicate(), timeout=timeout
        )
        return stdout.decode() or stderr.decode()
Prev
#Nanobot, Agent