"""容器内 fs 工具 helper(DockerExecutor 通过 `docker exec python tool_runner.py` 调用)。 调用约定: - argv[1] = tool name(read / write / edit / glob / grep) - stdin = JSON 序列化的 args(用 stdin 而非 argv 是为了不被 shell metachar 切路径, CJK / 引号 / 路径分隔符全透明传) - stdout = tool execute 返回的文本(LLM 拿到的) - exit code = 0 ok / 1 工具内部抛异常 / 2 参数 / unknown tool base_dir = `os.getcwd()` ── docker exec --workdir /workspace/ 已切到 task 工作目录 user_root = `/workspace` ── bind mount 边界,Tool._display 据此渲相对路径 不依赖任何 zcbot 自家包外的东西,纯用 `tools.fs` 五个 Tool 子类。容器镜像里 `/sandbox/tools/` 是 host repo `tools/` 目录的拷贝(Dockerfile `COPY tools/`)。 """ from __future__ import annotations import json import os import sys import traceback from pathlib import Path # 镜像里 /sandbox/ 下放了 tools/ 的拷贝,让 import 走 /sandbox/ sys.path.insert(0, "/sandbox") from tools.fs import EditTool, GlobTool, GrepTool, ReadTool, WriteTool # noqa: E402 TOOLS = { "read": ReadTool, "write": WriteTool, "edit": EditTool, "glob": GlobTool, "grep": GrepTool, } def main() -> int: if len(sys.argv) < 2: print("[Error] tool_runner: missing tool name argv[1]", file=sys.stderr) return 2 name = sys.argv[1] if name not in TOOLS: print(f"[Error] tool_runner: unknown tool: {name}", file=sys.stderr) return 2 try: args_raw = sys.stdin.read() args = json.loads(args_raw) if args_raw.strip() else {} except json.JSONDecodeError as e: print(f"[Error] tool_runner: invalid JSON args: {e}", file=sys.stderr) return 2 cls = TOOLS[name] # 缺必填参数早返清晰错误,而非让 execute 抛暴露内部签名的 TypeError(空 args 多由 # 上游 tool_call 损坏导致,AgentLoop 已重试兜底;漏到这里给模型一句能照做的话)。 required = (getattr(cls, "parameters", {}) or {}).get("required") or [] missing = [k for k in required if k not in args] if missing: print( f"[Error] bad arguments to {name}: 缺少必填参数 {missing};" f"请带齐 {required} 重新调用", file=sys.stderr, ) return 2 tool = cls(base_dir=Path(os.getcwd()), user_root=Path("/workspace")) try: result = tool.execute(**args) except TypeError as e: print(f"[Error] bad arguments to {name}: {e}", file=sys.stderr) return 2 except Exception as e: # 容器内 Tool 抛异常(IO / 权限等)── 落 stderr + 退非 0,DockerExecutor # 兜底成 ToolResult content;traceback 限 80 行防爆 LLM context print(f"[Error executing {name}] {type(e).__name__}: {e}", file=sys.stderr) tb = traceback.format_exc().splitlines() if len(tb) > 80: tb = tb[:40] + [f"... ({len(tb) - 40} lines truncated) ..."] print("\n".join(tb), file=sys.stderr) return 1 if not isinstance(result, str): result = str(result) sys.stdout.write(result) return 0 if __name__ == "__main__": sys.exit(main())