agent_bot/tools.py

import json
import urllib.request
import urllib.parse
import urllib.error
from html.parser import HTMLParser

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, mcp_api_url="http://localhost:8000"):
        self.mcp_api_url = mcp_api_url

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

        # 获取所有MCP服务的实际工具
        try:
            req = urllib.request.Request(f"{self.mcp_api_url}/tools")
            with urllib.request.urlopen(req) as response:
                mcp_tools = json.loads(response.read().decode('utf-8'))
            
            # 为每个MCP工具创建独立的工具定义
            for tool in mcp_tools:
                tools.append({
                    "type": "function",
                    "function": {
                        "name": tool.get("name"),  # 工具名称已经带有_mcp后缀
                        "description": tool.get("description", "未知工具"),
                        "parameters": tool.get("inputSchema", {
                            "type": "object",
                            "properties": {},
                            "required": []
                        })
                    }
                })
        except Exception as e:
            print(f"获取MCP工具列表时出错: {e}")
        
        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"):
            # 调用MCP API服务器
            try:
                # 构造请求数据
                data = {
                    "name": tool_name,
                    "arguments": parameters
                }
                
                # 发送POST请求到MCP API服务器
                req = urllib.request.Request(
                    f"{self.mcp_api_url}/call",
                    data=json.dumps(data).encode('utf-8'),
                    headers={'Content-Type': 'application/json'}
                )
                
                with urllib.request.urlopen(req) as response:
                    result = json.loads(response.read().decode('utf-8'))
                
                # 检查是否有错误
                if "error" in result:
                    error_msg = result['error']
                    return f"MCP调用错误: {error_msg}"
                
                # 返回结果
                return json.dumps(result, ensure_ascii=False)
            except urllib.error.HTTPError as e:
                return f"MCP调用HTTP错误: {e.code} - {e.reason}"
            except Exception as e:
                return f"MCP调用失败: {str(e)}"
        else:
            return f"未知工具: {tool_name}"

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