82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
"""
|
||
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}") |