"""HostExecutor:in-process 工具调用,沿用原 `Tool.execute` 行为。 用途: - 本地 dogfood / 单租户 / Step 1 默认 backend - Step 3 docker backend 引入后,承担"信任域 = host"那一半(read/write/edit/glob/ grep/load_skill/web_*/seedream/seedance,§7.5 #6),DockerExecutor 内部组合本类 + docker exec 处理 shell/run_python。 行为兼容性:错误分支与原 `AgentLoop._execute_tool_call` 三段(unknown tool / bad args / 抛异常)语义对齐 —— 都包成 `[Error] ...` content 返回,exit_code 区分内部用。 """ from __future__ import annotations from typing import Any, Dict, List from .executor import ExecCtx, Executor, ToolResult from tools.base import Tool class HostExecutor(Executor): def __init__(self, tools: Dict[str, Tool]) -> None: self._tools = tools def has_tool(self, name: str) -> bool: return name in self._tools def schemas(self) -> List[Dict[str, Any]]: return [t.schema for t in self._tools.values()] def call_tool(self, name: str, args: Dict[str, Any], ctx: ExecCtx) -> ToolResult: tool = self._tools.get(name) if tool is None: return ToolResult(content=f"[Error] unknown tool: {name}", exit_code=2) # 缺必填参数早返清晰错误,而非让 execute 抛 `missing N required positional # arguments` 这种暴露内部签名的 TypeError(空 args 多由上游 tool_call 损坏导致, # _stream_llm 已重试兜底;漏到这里就给模型一句能照做的话)。 required = (getattr(tool, "parameters", {}) or {}).get("required") or [] missing = [k for k in required if k not in args] if missing: return ToolResult( content=f"[Error] bad arguments to {name}: 缺少必填参数 {missing};" f"请带齐 {required} 重新调用", exit_code=2, ) try: result = tool.execute(**args) except TypeError as e: return ToolResult(content=f"[Error] bad arguments to {name}: {e}", exit_code=2) except Exception as e: return ToolResult( content=f"[Error executing {name}] {type(e).__name__}: {e}", exit_code=1, ) if not isinstance(result, str): result = str(result) return ToolResult(content=result, exit_code=0)