---

关于作者

• 深耕领域：大语言模型开发 / RAG 知识库 / AI Agent 落地 / 模型微调
• 技术栈：Python | RAG (LangChain / Dify + Milvus) | FastAPI + Docker
• 工程能力：专注模型工程化部署、知识库构建与优化，擅长全流程解决方案

「让 AI 交互更智能，让技术落地更高效」
欢迎技术探讨与项目合作，解锁大模型与智能交互的无限可能！

---

nanobot 工具系统：可扩展的能力框架

工具系统是 nanobot 与外部世界交互的桥梁，通过统一的抽象接口和注册机制，实现了高度可扩展的能力框架。

概述

nanobot 的工具系统采用"抽象基类 + 注册表"的设计模式，为 Agent 提供了与外部环境交互的能力。每个工具都实现了统一的接口，通过 ToolRegistry 进行管理和调度。本文将深入剖析工具系统的设计与实现。

问题背景

工具系统的核心挑战

构建一个灵活的工具系统需要解决以下问题：

挑战	描述	nanobot 的解决方案
接口统一	不同工具如何统一调用	Tool 抽象基类
参数验证	如何确保参数正确性	JSON Schema 验证
安全控制	如何防止危险操作	路径限制 + 命令过滤
动态扩展	如何动态添加工具	ToolRegistry 注册机制
错误处理	如何优雅处理执行错误	统一异常捕获

工具系统设计目标

核心架构

Tool 抽象基类

所有工具都继承自 Tool 抽象基类：

class Tool(ABC):
    """
    Abstract base class for agent tools.
    
    Tools are capabilities that the agent can use to interact with
    the environment, such as reading files, executing commands, etc.
    """
    
    _TYPE_MAP = {
        "string": str,
        "integer": int,
        "number": (int, float),
        "boolean": bool,
        "array": list,
        "object": dict,
    }
    
    @property
    @abstractmethod
    def name(self) -> str:
        """Tool name used in function calls."""
        pass
    
    @property
    @abstractmethod
    def description(self) -> str:
        """Description of what the tool does."""
        pass
    
    @property
    @abstractmethod
    def parameters(self) -> dict[str, Any]:
        """JSON Schema for tool parameters."""
        pass
    
    @abstractmethod
    async def execute(self, **kwargs: Any) -> str:
        """Execute the tool with given parameters."""
        pass

工具注册表

ToolRegistry 负责管理所有注册的工具：

class ToolRegistry:
    """
    Registry for agent tools.
    
    Allows dynamic registration and execution of tools.
    """
    
    def __init__(self):
        self._tools: dict[str, Tool] = {}
    
    def register(self, tool: Tool) -> None:
        """Register a tool."""
        self._tools[tool.name] = tool
    
    def get(self, name: str) -> Tool | None:
        """Get a tool by name."""
        return self._tools.get(name)
    
    def get_definitions(self) -> list[dict[str, Any]]:
        """Get all tool definitions in OpenAI format."""
        return [tool.to_schema() for tool in self._tools.values()]
    
    async def execute(self, name: str, params: dict[str, Any]) -> str:
        """Execute a tool by name with given parameters."""
        tool = self._tools.get(name)
        if not tool:
            return f"Error: Tool '{name}' not found"

        try:
            errors = tool.validate_params(params)
            if errors:
                return f"Error: Invalid parameters for tool '{name}': " + "; ".join(errors)
            return await tool.execute(**params)
        except Exception as e:
            return f"Error executing {name}: {str(e)}"

工具系统架构图

内置工具详解

文件操作工具

ReadFileTool

class ReadFileTool(Tool):
    """Tool to read file contents."""
    
    def __init__(self, allowed_dir: Path | None = None):
        self._allowed_dir = allowed_dir

    @property
    def name(self) -> str:
        return "read_file"
    
    @property
    def description(self) -> str:
        return "Read the contents of a file at the given path."
    
    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "The file path to read"
                }
            },
            "required": ["path"]
        }
    
    async def execute(self, path: str, **kwargs: Any) -> str:
        try:
            file_path = _resolve_path(path, self._allowed_dir)
            if not file_path.exists():
                return f"Error: File not found: {path}"
            if not file_path.is_file():
                return f"Error: Not a file: {path}"
            
            content = file_path.read_text(encoding="utf-8")
            return content
        except PermissionError as e:
            return f"Error: {e}"

WriteFileTool

class WriteFileTool(Tool):
    """Tool to write content to a file."""
    
    @property
    def name(self) -> str:
        return "write_file"
    
    @property
    def description(self) -> str:
        return "Write content to a file at the given path. Creates parent directories if needed."
    
    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "The file path to write to"
                },
                "content": {
                    "type": "string",
                    "description": "The content to write"
                }
            },
            "required": ["path", "content"]
        }

EditFileTool

class EditFileTool(Tool):
    """Tool to edit a file by replacing text."""
    
    @property
    def description(self) -> str:
        return "Edit a file by replacing old_text with new_text. The old_text must exist exactly in the file."
    
    async def execute(self, path: str, old_text: str, new_text: str, **kwargs: Any) -> str:
        try:
            file_path = _resolve_path(path, self._allowed_dir)
            content = file_path.read_text(encoding="utf-8")
            
            if old_text not in content:
                return f"Error: old_text not found in file. Make sure it matches exactly."
            
            # Count occurrences
            count = content.count(old_text)
            if count > 1:
                return f"Warning: old_text appears {count} times. Please provide more context to make it unique."
            
            new_content = content.replace(old_text, new_text, 1)
            file_path.write_text(new_content, encoding="utf-8")
            
            return f"Successfully edited {path}"

Shell 执行工具

ExecTool 是最强大但也最需要安全控制的工具：

class ExecTool(Tool):
    """Tool to execute shell commands."""
    
    def __init__(
        self,
        timeout: int = 60,
        working_dir: str | None = None,
        deny_patterns: list[str] | None = None,
        allow_patterns: list[str] | None = None,
        restrict_to_workspace: bool = False,
    ):
        self.timeout = timeout
        self.working_dir = working_dir
        self.deny_patterns = deny_patterns or [
            r"\brm\s+-[rf]{1,2}\b",          # rm -r, rm -rf
            r"\bdel\s+/[fq]\b",              # del /f, del /q
            r"\brmdir\s+/s\b",               # rmdir /s
            r"\b(format|mkfs|diskpart)\b",   # disk operations
            r"\bdd\s+if=",                   # dd
            r">\s*/dev/sd",                  # write to disk
            r"\b(shutdown|reboot|poweroff)\b",  # system power
            r":\(\)\s*\{.*\};\s*:",          # fork bomb
        ]

安全防护机制

网络工具

WebSearchTool

class WebSearchTool(Tool):
    """Search the web using Brave Search API."""
    
    name = "web_search"
    description = "Search the web. Returns titles, URLs, and snippets."
    parameters = {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
            "count": {"type": "integer", "description": "Results (1-10)", "minimum": 1, "maximum": 10}
        },
        "required": ["query"]
    }
    
    async def execute(self, query: str, count: int | None = None, **kwargs: Any) -> str:
        if not self.api_key:
            return "Error: BRAVE_API_KEY not configured"
        
        async with httpx.AsyncClient() as client:
            r = await client.get(
                "https://api.search.brave.com/res/v1/web/search",
                params={"q": query, "count": n},
                headers={"Accept": "application/json", "X-Subscription-Token": self.api_key},
                timeout=10.0
            )
        
        results = r.json().get("web", {}).get("results", [])
        # ... 格式化结果

WebFetchTool

class WebFetchTool(Tool):
    """Fetch and extract content from a URL using Readability."""
    
    name = "web_fetch"
    description = "Fetch URL and extract readable content (HTML to markdown/text)."
    
    async def execute(self, url: str, extractMode: str = "markdown", **kwargs: Any) -> str:
        from readability import Document
        
        # URL 验证
        is_valid, error_msg = _validate_url(url)
        if not is_valid:
            return json.dumps({"error": f"URL validation failed: {error_msg}"})
        
        async with httpx.AsyncClient(follow_redirects=True, max_redirects=5) as client:
            r = await client.get(url, headers={"User-Agent": USER_AGENT})
        
        # 根据内容类型处理
        if "application/json" in ctype:
            text = json.dumps(r.json(), indent=2)
        elif "text/html" in ctype:
            doc = Document(r.text)
            text = self._to_markdown(doc.summary())

参数验证机制

JSON Schema 验证

Tool 基类提供了内置的参数验证：

def validate_params(self, params: dict[str, Any]) -> list[str]:
    """Validate tool parameters against JSON schema. Returns error list (empty if valid)."""
    schema = self.parameters or {}
    if schema.get("type", "object") != "object":
        raise ValueError(f"Schema must be object type")
    return self._validate(params, {**schema, "type": "object"}, "")

def _validate(self, val: Any, schema: dict[str, Any], path: str) -> list[str]:
    t, label = schema.get("type"), path or "parameter"
    if t in self._TYPE_MAP and not isinstance(val, self._TYPE_MAP[t]):
        return [f"{label} should be {t}"]
    
    errors = []
    if "enum" in schema and val not in schema["enum"]:
        errors.append(f"{label} must be one of {schema['enum']}")
    if t in ("integer", "number"):
        if "minimum" in schema and val < schema["minimum"]:
            errors.append(f"{label} must be >= {schema['minimum']}")
        if "maximum" in schema and val > schema["maximum"]:
            errors.append(f"{label} must be <= {schema['maximum']}")
    # ... 更多验证规则
    return errors

OpenAI 格式转换

def to_schema(self) -> dict[str, Any]:
    """Convert tool to OpenAI function schema format."""
    return {
        "type": "function",
        "function": {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters,
        }
    }

工具注册流程

默认工具注册

在 AgentLoop 初始化时注册默认工具：

def _register_default_tools(self) -> None:
    """Register the default set of tools."""
    # 文件工具（可选限制到工作区）
    allowed_dir = self.workspace if self.restrict_to_workspace else None
    self.tools.register(ReadFileTool(allowed_dir=allowed_dir))
    self.tools.register(WriteFileTool(allowed_dir=allowed_dir))
    self.tools.register(EditFileTool(allowed_dir=allowed_dir))
    self.tools.register(ListDirTool(allowed_dir=allowed_dir))
    
    # Shell 工具
    self.tools.register(ExecTool(
        working_dir=str(self.workspace),
        timeout=self.exec_config.timeout,
        restrict_to_workspace=self.restrict_to_workspace,
    ))
    
    # 网络工具
    self.tools.register(WebSearchTool(api_key=self.brave_api_key))
    self.tools.register(WebFetchTool())
    
    # 消息工具
    message_tool = MessageTool(send_callback=self.bus.publish_outbound)
    self.tools.register(message_tool)
    
    # 子代理工具
    spawn_tool = SpawnTool(manager=self.subagents)
    self.tools.register(spawn_tool)
    
    # 定时任务工具
    if self.cron_service:
        self.tools.register(CronTool(self.cron_service))

扩展工具

创建自定义工具

from nanobot.agent.tools.base import Tool

class MyCustomTool(Tool):
    """A custom tool example."""
    
    @property
    def name(self) -> str:
        return "my_custom_tool"
    
    @property
    def description(self) -> str:
        return "Description of what this tool does."
    
    @property
    def parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "input": {
                    "type": "string",
                    "description": "Input parameter"
                }
            },
            "required": ["input"]
        }
    
    async def execute(self, input: str, **kwargs: Any) -> str:
        # 实现工具逻辑
        return f"Processed: {input}"

注册自定义工具

# 在 AgentLoop 中注册
self.tools.register(MyCustomTool())

工具执行流程

安全最佳实践

1. 路径限制

def _resolve_path(path: str, allowed_dir: Path | None = None) -> Path:
    """Resolve path and optionally enforce directory restriction."""
    resolved = Path(path).expanduser().resolve()
    if allowed_dir and not str(resolved).startswith(str(allowed_dir.resolve())):
        raise PermissionError(f"Path {path} is outside allowed directory")
    return resolved

2. 命令过滤

deny_patterns = [
    r"\brm\s+-[rf]{1,2}\b",          # rm -r, rm -rf
    r"\b(format|mkfs|diskpart)\b",   # disk operations
    r"\b(shutdown|reboot)\b",        # system power
]

3. 超时控制

try:
    stdout, stderr = await asyncio.wait_for(
        process.communicate(),
        timeout=self.timeout
    )
except asyncio.TimeoutError:
    process.kill()
    return f"Error: Command timed out after {self.timeout} seconds"

4. 输出截断

max_len = 10000
if len(result) > max_len:
    result = result[:max_len] + f"\n... (truncated, {len(result) - max_len} more chars)"

工具能力对比

工具	功能	安全措施
read_file	读取文件内容	路径限制
write_file	写入文件内容	路径限制、自动创建目录
edit_file	编辑文件内容	路径限制、唯一性检查
list_dir	列出目录内容	路径限制
exec	执行 Shell 命令	命令过滤、超时、路径限制
web_search	网络搜索	API Key 验证
web_fetch	获取网页内容	URL 验证、重定向限制
message	发送消息	渠道验证
spawn	创建子代理	资源隔离
cron	定时任务	权限控制

总结

nanobot 的工具系统通过统一的抽象接口和注册机制，实现了高度可扩展的能力框架：

设计原则	实现方式
统一接口	Tool 抽象基类定义 name、description、parameters、execute
参数验证	JSON Schema 验证，支持类型、范围、枚举等
安全控制	路径限制、命令过滤、超时控制、输出截断
动态扩展	ToolRegistry 注册机制，支持运行时添加工具
错误处理	统一异常捕获，返回友好的错误信息

工具系统是 nanobot 与外部世界交互的核心，其设计体现了"安全第一、易于扩展"的理念。

---