"""工具基类 - 工具系统重构 Phase 6.1""" from abc import ABC, abstractmethod from typing import Any, Generic, TypeVar from app.agents.tools.manifest import ( PermissionClass, SideEffectScope, ToolCategory, ToolManifest, ) T = TypeVar("T") class BaseTool(ABC, Generic[T]): """工具基类 提供工具的标准接口和默认实现。 所有自定义工具应继承此类。 """ def __init__( self, name: str, description: str, category: ToolCategory, permission_class: PermissionClass, side_effect_scope: SideEffectScope = SideEffectScope.NONE, requires_confirmation: bool = False, is_streaming: bool = False, tags: list[str] | None = None, ): self.name = name self.description = description self.category = category self.permission_class = permission_class self.side_effect_scope = side_effect_scope self.requires_confirmation = requires_confirmation self.is_streaming = is_streaming self.tags = tags or [] def get_manifest(self) -> ToolManifest: """获取工具元数据 Returns: 工具元数据 """ return ToolManifest( name=self.name, description=self.description, category=self.category, parameters=self.get_parameters(), return_schema=self.get_return_schema(), permission_class=self.permission_class, side_effect_scope=self.side_effect_scope, requires_confirmation=self.requires_confirmation, is_streaming=self.is_streaming, tags=self.tags, ) @abstractmethod def get_parameters(self) -> dict[str, Any]: """获取参数 Schema(JSON Schema 格式) Returns: 参数 schema """ pass @abstractmethod def get_return_schema(self) -> dict[str, Any]: """获取返回值 Schema Returns: 返回值 schema """ pass @abstractmethod async def execute(self, **kwargs) -> T: """执行工具 Args: **kwargs: 工具参数 Returns: 执行结果 """ pass async def execute_safe(self, **kwargs) -> dict[str, Any]: """安全执行工具,捕获异常 Args: **kwargs: 工具参数 Returns: 包含 success 和 result/error 的字典 """ try: result = await self.execute(**kwargs) return {"success": True, "result": result} except Exception as e: return {"success": False, "error": str(e)} def __repr__(self) -> str: return f"<{self.__class__.__name__}(name={self.name!r})>" class ReadTool(BaseTool): """只读工具基类""" def __init__(self, **kwargs): kwargs.setdefault("category", ToolCategory.READ) kwargs.setdefault("permission_class", PermissionClass.READ) kwargs.setdefault("side_effect_scope", SideEffectScope.NONE) super().__init__(**kwargs) class WriteTool(BaseTool): """写入工具基类""" def __init__(self, **kwargs): kwargs.setdefault("category", ToolCategory.WRITE) kwargs.setdefault("permission_class", PermissionClass.WRITE) kwargs.setdefault("side_effect_scope", SideEffectScope.LOCAL_STATE) super().__init__(**kwargs) class DBWriteTool(BaseTool): """数据库写入工具基类""" def __init__(self, **kwargs): kwargs.setdefault("category", ToolCategory.DB_WRITE) kwargs.setdefault("permission_class", PermissionClass.WRITE) kwargs.setdefault("side_effect_scope", SideEffectScope.DB_WRITE) kwargs.setdefault("requires_confirmation", True) super().__init__(**kwargs) class ExternalTool(BaseTool): """外部工具基类(执行外部命令等)""" def __init__(self, **kwargs): kwargs.setdefault("category", ToolCategory.EXTERNAL) kwargs.setdefault("permission_class", PermissionClass.EXTERNAL) kwargs.setdefault("side_effect_scope", SideEffectScope.NETWORK) kwargs.setdefault("requires_confirmation", True) super().__init__(**kwargs) class NetworkTool(BaseTool): """网络工具基类""" def __init__(self, **kwargs): kwargs.setdefault("category", ToolCategory.NETWORK) kwargs.setdefault("permission_class", PermissionClass.EXTERNAL) kwargs.setdefault("side_effect_scope", SideEffectScope.NETWORK) super().__init__(**kwargs)