"""单个 Agent Run 可见的工具视图。""" from __future__ import annotations from dataclasses import replace from typing import Any from app.application.tool_runtime.runtime import ToolRuntime from app.modules.tool_runtime.definition import ToolInput, ToolResult, ToolSpec class RunScopedToolView: """保存 Run 级工具别名;实际执行始终委托共享 ToolRuntime。""" def __init__(self, runtime: ToolRuntime) -> None: self._runtime = runtime self._aliases: dict[str, tuple[str, ToolSpec]] = {} def bind_alias(self, alias: str, target: str, *, description_prefix: str = "") -> bool: """将已有工具作为 Run 级别名暴露;目标不存在时返回 False。""" spec = next((item for item in self._runtime.list_specs() if item.name == target), None) if spec is None: return False description = f"{description_prefix}{spec.description}" if description_prefix else spec.description self._aliases[alias] = (target, replace(spec, name=alias, description=description)) return True def list_specs(self) -> list[ToolSpec]: return [*self._runtime.list_specs(), *(spec for _, spec in self._aliases.values())] def specs_by_name(self) -> dict[str, ToolSpec]: return {spec.name: spec for spec in self.list_specs()} async def execute( self, tool: str, input: ToolInput, *, run_id: str | None = None, character_id: str | None = None, trace_id: str | None = None, confirmed: bool = False, context: dict[str, Any] | None = None, ) -> ToolResult: target = self._aliases.get(tool, (tool,))[0] return await self._runtime.execute( target, input, run_id=run_id, character_id=character_id, trace_id=trace_id, confirmed=confirmed, context=context, ) __all__ = ["RunScopedToolView"]