+--------+      +-------+      +---------+|  User  | ---> |  LLM  | ---> |  Tool   || prompt |      |       |      | execute |+--------+      +---+---+      +----+----+                    ^                |                    |   tool_result  |                    +----------------+                    (loop until stop_reason != "tool_use")

+--------+      +-------+      +------------------+|  User  | ---> |  LLM  | ---> | Tool Dispatch    || prompt |      |       |      | {                |+--------+      +---+---+      |   bash: run_bash |                    ^           |   read: run_read |                    |           |   write: run_wr  |                    +-----------+   edit: run_edit |                    tool_result | }                |                                +------------------+The dispatch map is a dict: {tool_name: handler_function}.One lookup replaces any if/elif chain.

[用户问题] → [模型工具调用指令] → [工具执行结果]

    User Input        |        v    messages[] --> LLM API (tools=TOOLS)                       |                  stop_reason?                  /          \            "end_turn"    "tool_use"               |              |             Print    for each tool_use block:                        TOOL_HANDLERS[name](**input)                              |                        tool_result                              |                        messages[] <-- {role:"user", content:[tool_result]}                              |                        back to LLM --> may chain more tools                                          or "end_turn" --> Print

# Tool Usage NotesTool signatures are provided automatically via function calling.This file documents non-obvious constraints and usage patterns.## exec — Safety Limits- Commands have a configurable timeout (default 60s)- Dangerous commands are blocked (rm -rf, format, dd, shutdown, etc.)- Output is truncated at 10,000 characters- `restrictToWorkspace` config can limit file access to the workspace## cron — Scheduled Reminders- Please refer to cron skill for usage.

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.    """    # 定义JSON Schema类型到Python原生类型的映射字典，用于参数类型校验    # 核心作用：将Schema中定义的类型（如"string"）转换为Python可识别的类型（str），方便后续类型检查    _TYPE_MAP = {        "string": str,        # Schema字符串类型对应Python str        "integer": int,       # Schema整数类型对应Python int        "number": (int, float), # Schema数字类型对应Python int/float（兼容整数和浮点数）        "boolean": bool,      # Schema布尔类型对应Python bool        "array": list,        # Schema数组类型对应Python list        "object": dict,       # Schema对象类型对应Python dict    }    # 定义抽象属性：工具名称（必须由子类实现）    # 作用：指定工具在函数调用中的唯一标识（如"exec"），LLM通过该名称调用对应工具    @property    @abstractmethod    def name(self) -> str:        """Tool name used in function calls."""        pass    # 定义抽象属性：工具描述（必须由子类实现）    # 作用：向LLM说明工具的功能，帮助LLM判断何时调用该工具，描述需清晰易懂    @property    @abstractmethod    def description(self) -> str:        """Description of what the tool does."""        pass    # 定义抽象属性：工具参数Schema（必须由子类实现）    # 作用：定义工具入参的JSON Schema规范，用于参数校验和向LLM声明参数格式    @property    @abstractmethod    def parameters(self) -> dict[str, Any]:        """JSON Schema for tool parameters."""        pass    # 定义抽象方法：工具执行逻辑（必须由子类实现）    # 作用：实现工具的核心功能，接收参数并返回执行结果，async标识异步执行（适配Agent异步架构）    @abstractmethod    async def execute(self, **kwargs: Any) -> str:        """        Execute the tool with given parameters.        Args:            **kwargs: Tool-specific parameters.        Returns:            String result of the tool execution.        """        pass    # 工具参数校验入口方法：校验传入参数是否符合当前工具的Schema规范    # 返回值：校验错误信息列表（空列表表示校验通过）    def validate_params(self, params: dict[str, Any]) -> list[str]:        """Validate tool parameters against JSON schema. Returns error list (empty if valid)."""        # 获取当前工具的参数Schema，若未定义则默认为空字典        schema = self.parameters or {}        # 校验Schema的顶层类型必须是"object"（因为参数本质是键值对），否则抛出异常        if schema.get("type", "object") != "object":            raise ValueError(f"Schema must be object type, got {schema.get('type')!r}")        # 调用内部递归校验方法，传入待校验参数、完整Schema、空路径（用于定位错误参数）        return self._validate(params, {**schema, "type": "object"}, "")    # 内部递归校验方法：核心参数校验逻辑，支持嵌套类型（如对象、数组）的递归校验    # val：待校验的参数值；schema：当前层级的Schema规范；path：参数的路径（用于精准定位错误，如"command.working_dir"）    def _validate(self, val: Any, schema: dict[str, Any], path: str) -> list[str]:        # 获取当前Schema定义的类型和参数路径标签（用于错误提示）        t, label = schema.get("type"), path or "parameter"        # 第一步：基础类型校验——若Schema类型在映射表中，检查参数值是否为对应Python类型        if t in self._TYPE_MAP and not isinstance(val, self._TYPE_MAP[t]):            return [f"{label} should be {t}"]        # 初始化错误列表，用于收集所有校验错误        errors = []        # 第二步：枚举值校验——若Schema定义了enum，检查参数值是否在枚举列表中        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"):            # 最小值校验：若定义了minimum，检查参数值是否大于等于最小值            if "minimum" in schema and val < schema["minimum"]:                errors.append(f"{label} must be >= {schema['minimum']}")            # 最大值校验：若定义了maximum，检查参数值是否小于等于最大值            if "maximum" in schema and val > schema["maximum"]:                errors.append(f"{label} must be <= {schema['maximum']}")        # 第四步：字符串类型长度校验        if t == "string":            # 最小长度校验：若定义了minLength，检查字符串长度是否达标            if "minLength" in schema and len(val) < schema["minLength"]:                errors.append(f"{label} must be at least {schema['minLength']} chars")            # 最大长度校验：若定义了maxLength，检查字符串长度是否超限            if "maxLength" in schema and len(val) > schema["maxLength"]:                errors.append(f"{label} must be at most {schema['maxLength']} chars")        # 第五步：对象类型校验（支持嵌套对象）        if t == "object":            # 获取对象的属性定义和必填属性列表            props = schema.get("properties", {})            # 必填属性校验：检查所有必填属性是否存在于参数中            for k in schema.get("required", []):                if k not in val:                    errors.append(f"missing required {path + '.' + k if path else k}")            # 递归校验对象的每个属性值：若属性在Schema中定义，则递归校验其值            for k, v in val.items():                if k in props:                    errors.extend(self._validate(v, props[k], path + '.' + k if path else k))        # 第六步：数组类型校验（支持数组元素的递归校验）        if t == "array" and "items" in schema:            # 遍历数组每个元素，递归校验元素是否符合items定义的Schema            for i, item in enumerate(val):                errors.extend(self._validate(item, schema["items"], f"{path}[{i}]" if path else f"[{i}]"))        # 返回所有校验错误        return errors    # 工具Schema转换方法：将当前工具转换为OpenAI函数调用的Schema格式    # 作用：让LLM能识别工具的调用格式，兼容OpenAI Function Call生态    def to_schema(self) -> dict[str, Any]:        """Convert tool to OpenAI function schema format."""        return {            "type": "function",            "function": {                "name": self.name,          # 工具名称（对应OpenAI函数名）                "description": self.description, # 工具描述（帮助LLM理解工具用途）                "parameters": self.parameters,   # 工具参数Schema（声明入参格式）            }        }

用户请求─ 包含搜索/查找意图？    ─ 是        ─ 涉及文件操作？            ─ 是 → 使用 exec 工具执行 grep            ─ 否 → 继续判断    ─ 否 → 判断其他工具需求

"""Shell execution tool."""class ExecTool(Tool):    """Tool to execute shell commands."""        # 工具初始化方法：配置Shell执行的安全参数和运行参数    # 参数说明：    # - timeout：命令执行超时时间（默认60秒）    # - working_dir：默认工作目录（若未指定则使用当前目录）    # - deny_patterns：危险命令正则黑名单（默认内置常见危险命令）    # - allow_patterns：命令允许列表（白名单，为空则不启用）    # - restrict_to_workspace：是否限制命令仅能访问指定工作目录（防止越权访问）    # - path_append：追加到环境变量PATH的路径（用于指定命令查找路径）    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,        path_append: str = "",    ):        self.timeout = timeout  # 初始化命令超时时间        self.working_dir = working_dir  # 初始化默认工作目录        # 初始化危险命令黑名单：默认拦截删除、格式化、系统关机、fork炸弹等破坏性命令        self.deny_patterns = deny_patterns or [            r"\brm\s+-[rf]{1,2}\b",          # 匹配rm -r/rm -rf/rm -fr（递归删除文件）            r"\bdel\s+/[fq]\b",              # 匹配del /f/del /q（强制/静默删除文件，Windows）            r"\brmdir\s+/s\b",               # 匹配rmdir /s（删除目录及子目录，Windows）            r"(?:^|[;&|]\s*)format\b",       # 匹配format命令（格式化磁盘，仅匹配独立命令）            r"\b(mkfs|diskpart)\b",          # 匹配磁盘操作命令（mkfs格式化文件系统、diskpart磁盘分区）            r"\bdd\s+if=",                   # 匹配dd命令（磁盘写入，if指定输入文件）            r">\s*/dev/sd",                  # 匹配写入磁盘设备（如> /dev/sda，破坏性操作）            r"\b(shutdown|reboot|poweroff)\b",  # 匹配系统关机/重启/断电命令            r":\(\)\s*\{.*\};\s*:",          # 匹配fork炸弹（无限创建进程，导致系统崩溃）        ]        self.allow_patterns = allow_patterns or []  # 初始化命令允许列表（默认空，不启用）        self.restrict_to_workspace = restrict_to_workspace  # 初始化工作目录限制开关        self.path_append = path_append  # 初始化PATH环境变量追加路径    # 实现抽象属性：工具名称（固定为"exec"，LLM通过该名称调用）    @property    def name(self) -> str:        return "exec"    # 实现抽象属性：工具描述（说明工具功能并提醒谨慎使用）    @property    def description(self) -> str:        return "Execute a shell command and return its output. Use with caution."    # 实现抽象属性：工具参数Schema（定义exec工具的入参规范）    @property    def parameters(self) -> dict[str, Any]:        return {            "type": "object",  # 顶层类型为对象（键值对）            "properties": {                # 命令参数：必填，字符串类型，描述要执行的Shell命令                "command": {                    "type": "string",                    "description": "The shell command to execute"                },                # 工作目录参数：可选，字符串类型，描述命令执行的工作目录                "working_dir": {                    "type": "string",                    "description": "Optional working directory for the command"                }            },            "required": ["command"]  # 声明command为必填参数        }    # 实现抽象方法：工具核心执行逻辑（异步执行Shell命令）    # 参数：command（要执行的命令）、working_dir（临时工作目录）、其他扩展参数    async def execute(self, command: str, working_dir: str | None = None, **kwargs: Any) -> str:        # 确定命令执行的工作目录：优先使用传入的working_dir，其次是工具默认的working_dir，最后是当前目录        cwd = working_dir or self.working_dir or os.getcwd()        # 执行命令安全校验：检查命令是否包含危险内容、是否越权访问路径等        guard_error = self._guard_command(command, cwd)        # 若安全校验失败，直接返回错误信息（不执行命令）        if guard_error:            return guard_error        # 复制当前环境变量（避免修改全局环境变量）        env = os.environ.copy()        # 若配置了PATH追加路径，将其添加到环境变量PATH中        if self.path_append:            env["PATH"] = env.get("PATH", "") + os.pathsep + self.path_append        try:            # 异步创建Shell子进程执行命令：            # - stdout/stderr重定向到管道（用于捕获输出）            # - cwd指定工作目录            # - env指定环境变量            process = await asyncio.create_subprocess_shell(                command,                stdout=asyncio.subprocess.PIPE,                stderr=asyncio.subprocess.PIPE,                cwd=cwd,                env=env,            )            try:                # 等待命令执行完成并捕获输出，设置超时时间（防止命令挂起）                stdout, stderr = await asyncio.wait_for(                    process.communicate(),                    timeout=self.timeout                )            except asyncio.TimeoutError:                # 命令超时：终止进程                process.kill()                # 等待进程完全终止（释放文件描述符等资源）                try:                    await asyncio.wait_for(process.wait(), timeout=5.0)                except asyncio.TimeoutError:                    pass                # 返回超时错误信息                return f"Error: Command timed out after {self.timeout} seconds"            # 初始化输出列表，用于拼接标准输出、标准错误、退出码            output_parts = []            # 若有标准输出，解码为字符串（utf-8，无法解码的字符替换）并添加到输出列表            if stdout:                output_parts.append(stdout.decode("utf-8", errors="replace"))            # 若有标准错误，解码后添加到输出列表（标注STDERR）            if stderr:                stderr_text = stderr.decode("utf-8", errors="replace")                if stderr_text.strip():  # 仅当标准错误非空时添加                    output_parts.append(f"STDERR:\n{stderr_text}")            # 若命令退出码非0（执行失败），添加退出码信息到输出列表            if process.returncode != 0:                output_parts.append(f"\nExit code: {process.returncode}")            # 拼接所有输出部分，若无输出则返回"(no output)"            result = "\n".join(output_parts) if output_parts else "(no output)"            # 截断超长输出：限制最大长度为10000字符，避免返回结果过大            max_len = 10000            if len(result) > max_len:                result = result[:max_len] + f"\n... (truncated, {len(result) - max_len} more chars)"            # 返回命令执行结果            return result        except Exception as e:            # 捕获其他执行异常（如进程创建失败），返回错误信息            return f"Error executing command: {str(e)}"    # 命令安全防护方法：校验命令是否安全，返回错误信息（None表示安全）    # 核心作用：拦截危险命令、越权路径访问，是ExecTool的核心安全机制    def _guard_command(self, command: str, cwd: str) -> str | None:        """Best-effort safety guard for potentially destructive commands."""        # 去除命令首尾空格，便于正则匹配        cmd = command.strip()        # 转换为小写（正则匹配不区分大小写）        lower = cmd.lower()        # 第一步：黑名单校验——检查命令是否匹配危险模式        for pattern in self.deny_patterns:            if re.search(pattern, lower):                return "Error: Command blocked by safety guard (dangerous pattern detected)"        # 第二步：白名单校验——若配置了允许列表，命令必须匹配其中一个模式        if self.allow_patterns:            if not any(re.search(p, lower) for p in self.allow_patterns):                return "Error: Command blocked by safety guard (not in allowlist)"        # 第三步：工作目录限制校验——若开启限制，拦截路径遍历和越权访问        if self.restrict_to_workspace:            # 拦截路径遍历字符（../ 或 ..\），防止访问上级目录            if "..\\" in cmd or "../" in cmd:                return "Error: Command blocked by safety guard (path traversal detected)"            # 解析工作目录的绝对路径（用于后续路径校验）            cwd_path = Path(cwd).resolve()            # 提取命令中的Windows绝对路径（如C:\Users\test）            win_paths = re.findall(r"[A-Za-z]:\\[^\\\"']+", cmd)            # 提取命令中的POSIX绝对路径（如/usr/bin），仅匹配独立的绝对路径（避免误匹配）            posix_paths = re.findall(r"(?:^|[\s|>])(/[^\s\"'>]+)", cmd)            # 遍历所有提取的绝对路径，检查是否超出工作目录范围            for raw in win_paths + posix_paths:                try:                    # 解析路径为绝对路径                    p = Path(raw.strip()).resolve()                except Exception:                    # 路径解析失败（如非法路径），跳过校验                    continue                # 若路径是绝对路径且不在工作目录及其子目录中，拦截命令                if p.is_absolute() and cwd_path not in p.parents and p != cwd_path:                    return "Error: Command blocked by safety guard (path outside working dir)"        # 所有校验通过，返回None（命令安全）        return None

3.5 沙箱

3.5.1 两层实现

nanobot的沙箱实现分两层：

第一层：命令守卫(软防护)。ExecTool._guard_command()在执行前用正则表达式检查命令：

• deny_patterns：默认屏蔽rm-rf、dd、format、shutdown、forkbomb等危险模式
• allow_patterns：可选白名单，只有匹配的命令才放行
• restrict_to_workspace：若开启，绝对路径必须在workspace目录内(防止路径逃逸)
• 内网URL检测：调用security.network.contains_internal_url() 屏蔽访问内网地址
• 保护内部文件：屏蔽直接写入history.jsonl/.dream_cursor(防止LLM篡改记忆)

第二层：bubblewrap(硬隔离，Linux 容器)。sandbox.py的  _bwrap()把命令包裹在bwrap 沙箱里：

• 只读挂载/usr、/bin、系统库
• 用tmpfs隐藏workspace 的父目录(config.json 所在目录)
• workspace读写挂载，media目录只读挂载
• 新进程组(–new-session)、进程死亡传播(–die-with-parent)

环境变量隔离_build_env()只传递最小环境(HOME/LANG/TERM)，屏蔽所有API key等敏感变量(allowed_env_keys 白名单除外)。

3.5.2 流程

我们加入 沙箱之后，重新走以下完整的运行流程，从接受到用户消息开始，一直到最后。

3.5.3 代码

bwrap(bubblewrap)是Linux的非特权容器工具，nanobot的_bwrap()函数用它构造如下命令：

bubblewrap的实现如下：

def_bwrap(command: str, workspace: str, cwd: str) -> str:"""将命令包裹在 bubblewrap 沙箱中执行 (需要容器内存在 bwrap)。    只有 workspace 目录以读写方式挂载；其父目录(存放 config.json 的位置)    被一个全新的 tmpfs 覆盖隐藏。media 目录以只读方式挂载，供 exec 命令    读取用户上传的附件。    安全属性:      - config.json (含 API key) 在沙箱内不可见      - workspace 的父目录整体被替换为空 tmpfs。      - 文件系统写操作被限制在 workspace 内；其他路径均为只读或不可见。      - 进程运行在新会话 (--new-session) 中，外部 TTY 的信号无法传入；        父进程退出时沙箱进程自动被回收 (--die-with-parent)，防止孤儿进程。    """    ws = Path(workspace).resolve()    media = get_media_dir().resolve()# 确保 cwd 始终在沙箱内。如果传入的路径已逃出 workspace 根目录# (例如传了绝对路径)，则回退到 workspace 根，防止 bwrap 收到# 沙箱外的 --chdir 参数。try:        sandbox_cwd = str(ws / Path(cwd).resolve().relative_to(ws))except ValueError:        sandbox_cwd = str(ws)# 必须存在于宿主机的路径 (bind 挂载失败会报错)。    required = ["/usr"]# 可选路径；--ro-bind-try 在路径不存在时静默跳过。    optional = ["/bin", "/lib", "/lib64", "/etc/alternatives","/etc/ssl/certs", "/etc/resolv.conf", "/etc/ld.so.cache"]    args = ["bwrap","--new-session", # 脱离调用方的 TTY/会话；防止 SIGHUP、SIGINT# 等信号从宿主机泄漏到沙箱进程。"--die-with-parent",  # nanobot 退出时自动回收沙箱进程, 防止孤儿进程残留。    ]# 挂载运行 shell 和常用 CLI 工具所需的最小只读系统目录树。for p in required:        args += ["--ro-bind", p, p]for p in optional:        args += ["--ro-bind-try", p, p]  # 路径不存在时静默跳过。    args += ["--proc", "/proc",     # 许多工具 (ps、top、/proc/self/...) 依赖 /proc。"--dev", "/dev",       # /dev/null、/dev/urandom 等设备节点所必需。"--tmpfs", "/tmp",     # 隔离的 /tmp, 与宿主机不共享。# 将 workspace 的*父目录*替换为空 tmpfs。# 这会隐藏 config.json (含 API key) 以及同级目录# (如 ~/.nanobot/sessions、~/.nanobot/media 的上层),# 使沙箱内进程无法访问这些文件。"--tmpfs", str(ws.parent),# 在刚刚创建的空 tmpfs 下重建 workspace 挂载点,# 再将真实的 workspace 目录以读写方式绑定挂载进去。# LLM 可以在 workspace 内自由读写文件。"--dir", str(ws),"--bind", str(ws), str(ws),# 将 media 上传目录以只读方式暴露给沙箱,# 使 agent 可以读取用户附件, 但无法修改或删除。"--ro-bind-try", str(media), str(media),"--chdir", sandbox_cwd,    # 在正确的工作目录中启动命令。"--", "sh", "-c", command,    ]return shlex.join(args)        

3.6 并行 vs 串行

nanoboot 中，工具默认是串行 (sequential)执行。并行是可选特性, 且受工具属性限制。

即使开启, 也不是”所有工具一起跑”，而是按**批次(batch)**执行，批次内并行，批次间串行。

3.6.1 并行执行规则

并行执行规则 (_partition_tool_batches)如下：

3.6.2 并行属性

各个工具的并行属性如下：

继续打广告^_^