"""Tool 抽象基类 - 统一工具接口""" from abc import ABC, abstractmethod from typing import Any class Tool(ABC): """工具抽象基类 所有工具(FunctionTool, AgentTool, MCPTool)的统一接口。 """ def __init__( self, name: str, description: str, input_schema: dict[str, Any] | None = None, output_schema: dict[str, Any] | None = None, version: str = "1.0.0", tags: list[str] | None = None, ): self.name = name self.description = description self.input_schema = input_schema self.output_schema = output_schema self.version = version self.tags = tags or [] @abstractmethod async def execute(self, **kwargs) -> dict: """执行工具,返回结果 dict""" ... async def before_execute(self, **kwargs) -> None: """执行前钩子""" pass async def after_execute(self, result: dict, **kwargs) -> None: """执行后钩子""" pass async def on_error(self, error: Exception, **kwargs) -> None: """错误钩子""" pass async def safe_execute(self, **kwargs) -> dict: """带钩子的安全执行""" try: await self.before_execute(**kwargs) result = await self.execute(**kwargs) await self.after_execute(result, **kwargs) return result except Exception as e: await self.on_error(e, **kwargs) raise def to_dict(self) -> dict: return { "name": self.name, "description": self.description, "input_schema": self.input_schema, "output_schema": self.output_schema, "version": self.version, "tags": self.tags, } def __repr__(self) -> str: return f"<{type(self).__name__} name={self.name!r} version={self.version}>"