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工具
        tools.extend(self.mcp_tools.get_mcp_tool_list())
        
        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工具
        elif tool_name.endswith("_mcp"):
            result = self.mcp_tools.call_mcp_tool(tool_name, parameters)
            # 解析结果并检查是否有错误
            try:
                result_dict = json.loads(result)
                if "error" in result_dict:
                    error_msg = result_dict['error']
                    # 提供更详细的错误信息和建议
                    if "unhandled errors in a TaskGroup" in error_msg:
                        server_name = tool_name[5:-4]  # 提取服务名
                        return f"MCP调用错误: {server_name}服务内部出现未处理的异常。这通常表示服务端存在问题。错误详情: {error_msg}"
                    return f"MCP调用错误: {error_msg}"
            except json.JSONDecodeError:
                pass  # 如果不是JSON格式，直接返回原始结果
            return result

        return f"工具 {tool_name} 执行完成"

if __name__ == "__main__":
    tools = Tools()
    print(tools.get_tool_list())
    print(tools.call_tool("read_webpage", {"url": "https://www.baidu.com"}))