agent_bot/tools.py

import json
import requests

def old_tool():
    """
    模拟器旧工具，固定返回100
    """
    return 100

class Tools:
    def __init__(self, mcp_api_url="http://localhost:8000"):
        self.mcp_api_url = mcp_api_url
        self.mcp_tools_cache = []

    def get_tool_list(self):

        # 首先从缓存中读取tools列表
        if self.mcp_tools_cache:
            return self.mcp_tools_cache

        # 基础工具,旧版tools
        tools = [
            {
                "type": "function",
                "function": {
                    "name": "luck_num",
                    "description": "获取幸运数字",
                    "parameters": {"type": "object", "properties": {"choice": {"type": "string", "description": "默认传入yes"}}}
                }
            }
        ]

        # 获取所有MCP服务的实际工具
        try:
            response = requests.get(f"{self.mcp_api_url}/tools")
            mcp_tools = response.json()
            
            # 为每个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}")

        # 存入缓存，便于下次快速调用
        self.mcp_tools_cache =  tools
        
        return tools

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

        # 处理原有的工具
        if tool_name == "luck_num":
            return str(old_tool())
        
        # 处理MCP工具
        elif tool_name.endswith("_mcp"):
            # 调用MCP API服务器
            try:
                # 构造请求数据
                data = {
                    "name": tool_name,
                    "arguments": parameters
                }
                
                # 发送POST请求到MCP API服务器
                response = requests.post(
                    f"{self.mcp_api_url}/call",
                    json=data  # 使用json参数自动设置Content-Type为application/json
                )
                response.raise_for_status()
                
                result = response.json()
                
                # 检查是否有错误
                if "error" in result:
                    error_msg = result['error']
                    return f"MCP调用错误: {error_msg}"
                
                # 返回结果
                return json.dumps(result, ensure_ascii=False)
            except requests.exceptions.RequestException as e:
                return f"MCP调用HTTP错误: {str(e)}"
            except Exception as e:
                return f"MCP调用失败: {str(e)}"
        else:
            return f"未知工具: {tool_name}"

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