agent_bot/tools.py

import json
import urllib.request
import urllib.parse
from html.parser import HTMLParser
from mcp_tools import MCPTools

class MLStripper(HTMLParser):
    def __init__(self):
        super().__init__()
        self.reset()
        self.fed = []
    
    def handle_data(self, d):
        self.fed.append(d)
    
    def get_data(self):
        return ''.join(self.fed)

def strip_tags(html):
    s = MLStripper()
    s.feed(html)
    return s.get_data()

class Tools:
    def __init__(self):
        self.mcp_tools = MCPTools()

    def get_tool_list(self):
        # 基础工具
        tools = [
            {
                "type": "function",
                "function": {
                    "name": "read_webpage",
                    "description": "读取指定URL网页的文本内容",
                    "parameters": {"type": "object", "properties": {"url": {"type": "string", "description": "要读取的网页URL"}}}
                }
            }
        ]

        # 获取所有MCP服务的实际工具
        mcp_tools = self.mcp_tools.get_all_mcp_tools_sync()
        
        # 为每个MCP工具创建独立的工具定义
        for tool in mcp_tools:
            tools.append({
                "type": "function",
                "function": {
                    "name": tool.get("name", "unknown_tool"),
                    "description": tool.get("description", "未知工具"),
                    "parameters": tool.get("inputSchema", {
                        "type": "object",
                        "properties": {},
                        "required": []
                    })
                }
            })
        
        return tools

    def call_tool(self, tool_name, parameters):
        print(f"🔧 正在执行工具: {tool_name}({parameters})")

        # 处理原有的工具
        if tool_name == "read_webpage":
            # 实现读取网页内容功能
            try:
                url = parameters["url"]
                req = urllib.request.Request(url, headers={
                    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
                })
                with urllib.request.urlopen(req) as response:
                    html = response.read().decode('utf-8')
                    text = strip_tags(html)
                    return text
            except Exception as e:
                return f"读取网页错误: {str(e)}"
        
        # 处理MCP工具
        else:
            # 根据工具名称确定所属的MCP服务
            server_name = self.mcp_tools.get_server_for_tool(tool_name)
            if server_name:
                # 构造正确的MCP调用参数
                mcp_params = {
                    "method": "tools/call",
                    "params": {
                        "name": tool_name,
                        "arguments": parameters
                    }
                }
                
                mcp_tool_name = f"call_{server_name}_mcp"
                result = self.mcp_tools.call_mcp_tool(mcp_tool_name, mcp_params)
                
                # 解析结果并检查是否有错误
                try:
                    result_dict = json.loads(result)
                    if "error" in result_dict:
                        error_msg = result_dict['error']
                        return f"MCP调用错误: {error_msg}"
                except json.JSONDecodeError:
                    pass  # 如果不是JSON格式，直接返回原始结果
                return result
            else:
                return f"未知工具: {tool_name}"

if __name__ == "__main__":
    tools = Tools()
    print(tools.get_tool_list())