"""Tool 基类: 子类只需声明 name/description/parameters 和 execute。""" from __future__ import annotations from abc import ABC, abstractmethod from pathlib import Path from typing import Optional # docker 沙箱把 user_root(`/users/`)bind 到容器内 `/workspace` # (agent_builder _build_system_prompt / executor_docker)。host-side 文件工具 # (send_email / wechat_push)在宿主进程跑,agent 给的可能是容器绝对路径 # `/workspace//x`,需翻回宿主 `user_root//x`。 _CONTAINER_ROOT = "/workspace" class FileOutOfBounds(Exception): """agent 给的文件路径解析后落在 user_root 之外(防 `../` 读到别人/系统文件)。""" def compact_tool_output( text: str, *, max_chars: int = 8_000, head_chars: int = 4_000, tail_chars: int = 2_000, ) -> str: """压缩长工具输出,保留头尾和截断说明。""" if len(text) <= max_chars: return text head_chars = max(0, min(head_chars, max_chars)) tail_chars = max(0, min(tail_chars, max_chars - head_chars)) removed = len(text) - head_chars - tail_chars return ( text[:head_chars] + f"\n[... truncated, {removed} chars omitted ...]\n" + (text[-tail_chars:] if tail_chars else "") ) class Tool(ABC): name: str = "" description: str = "" parameters: dict = {} def __init__(self, base_dir: Optional[Path] = None, user_root: Optional[Path] = None) -> None: self.base_dir: Path = Path(base_dir) if base_dir else Path.cwd() # tool 输出渲染路径用:user_root 内的 path 渲成相对 POSIX 串,user_root 外 # (用户 --working-dir 指向外部目录)保持绝对。None → 全部按绝对渲染。 # 目的:不让 tool result 文本里出现 user_id / 部署绝对路径,SPA 截图分享更安全; # 顺便让 web SPA 的 artifact chip 抽取(限定 / 前缀)更稳。 self.user_root: Optional[Path] = Path(user_root) if user_root else None @abstractmethod def execute(self, **kwargs) -> str: ... @property def schema(self) -> dict: return { "type": "function", "function": { "name": self.name, "description": self.description, "parameters": self.parameters, }, } def _resolve(self, path: str) -> Path: p = Path(path) return p if p.is_absolute() else (self.base_dir / p) def _resolve_user_file(self, raw: str) -> Path: """把 agent 给的路径解析成**宿主**绝对 Path,并强制落在 user_root 内。 host-side 文件工具(send_email / wechat_push)专用。处理三种入参: - 容器绝对路径 `/workspace//x` → 翻回宿主 `user_root//x`(docker bind); - 相对路径 `x` / `/x` → 拼到 `base_dir`(应为该 task 的宿主工作目录); - 宿主绝对路径 → 原样。 解析后越界(不在 user_root 下)抛 `FileOutOfBounds(raw)`。不校验存在性, 调用方按需 `.is_file()`。host 模式 agent 不会产出 `/workspace/` 前缀,翻译分支天然不触发。 """ raw = (raw or "").strip() if self.user_root is not None and ( raw == _CONTAINER_ROOT or raw.startswith(_CONTAINER_ROOT + "/") ): rest = raw[len(_CONTAINER_ROOT):].lstrip("/") p = self.user_root / rest else: p = self._resolve(raw) p = p.resolve() if self.user_root is not None: try: p.relative_to(self.user_root.resolve()) except ValueError: raise FileOutOfBounds(raw) return p def _display(self, p: Path) -> str: """对外渲染路径:在 user_root 内 → POSIX 相对串;否则原绝对。""" if self.user_root is not None: try: return p.resolve().relative_to(self.user_root.resolve()).as_posix() except (ValueError, OSError): pass return str(p)