""" tools/base_tool.py 所有工具的抽象基类,定义统一接口规范 """ from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any from mcp.mcp_protocol import ToolSchema from utils.logger import get_logger @dataclass class ToolResult: """工具执行结果的统一封装""" success: bool output: str metadata: dict[str, Any] = None def __post_init__(self): self.metadata = self.metadata or {} class BaseTool(ABC): """ 工具基类:所有具体工具必须继承此类并实现 execute() 方法 子类示例: class MyTool(BaseTool): name = "my_tool" description = "这是我的工具" parameters = {"input": {"type": "string", "description": "输入内容"}} def execute(self, **kwargs) -> ToolResult: return ToolResult(success=True, output=f"处理结果: {kwargs['input']}") """ # 子类必须定义的类属性 name: str = "" description: str = "" parameters: dict[str, Any] = {} def __init__(self): self.logger = get_logger("TOOL") @abstractmethod def execute(self, **kwargs) -> ToolResult: """ 执行工具逻辑(子类必须实现) Args: **kwargs: 工具参数,与 parameters 中定义的字段对应 Returns: ToolResult 实例 """ ... def get_schema(self) -> ToolSchema: """返回工具的 MCP Schema 描述""" return ToolSchema( name=self.name, description=self.description, parameters=self.parameters, ) def safe_execute(self, **kwargs) -> ToolResult: """ 带异常捕获的安全执行入口,由 MCP Server 调用 Returns: ToolResult,失败时 success=False 并携带错误信息 """ self.logger.info(f"▶ 执行工具 [{self.name}],参数: {kwargs}") try: result = self.execute(**kwargs) self.logger.info(f"✅ 工具 [{self.name}] 执行成功") return result except Exception as exc: self.logger.error(f"❌ 工具 [{self.name}] 执行失败: {exc}") return ToolResult(success=False, output=f"工具执行异常: {exc}")