LQAgentPlatform/function/load_mcp_server.py

import json
import logging
import os
from dotenv import load_dotenv
from typing import Any, Dict
from langchain_mcp_tools import convert_mcp_to_langchain_tools

from base.config import config_handler
from logger.loggering import server_logger

from utils.common import handler_err


class LoadMcpServer:

    def __init__(self):
        self.tools = []   # 默认为空列表
        self.cleanup = None
        self.mcp_server_config_path = config_handler.get("mcp","MCP_SERVER_CONFIG_PATH")
        # 加载MCP配置文件信息
        self.mcp_configs = self.load_mcp_configs(self.mcp_server_config_path)
    

    def load_mcp_configs(self, conf_file_path: str) -> Dict[str, Any]:
        """
            从 JSON 文件加载服务器配置
            Args:
                conf_file_path: JSON 配置文件路径
            Returns:
                包含服务器配置的字典
        """
          # 获取当前文件的目录
        current_dir = os.path.dirname(__file__)
        # 构建到 .env 的相对路径
        conf_file_path = os.path.join(current_dir , '../',  'config' , conf_file_path)
        server_logger.info(f"加载服务器配置: {conf_file_path}")
        if not os.path.exists(conf_file_path):
            server_logger.error(f"配置文件不存在: {conf_file_path}")
            raise ValueError(f"'conf_file_path'：{conf_file_path} 不存在！！")
        try:
            with open(conf_file_path, "r") as f:
                return json.load(f)


        except Exception as e:
            server_logger.error(f"❌ 加载 MCP 配置文件失败: {e}")


    async def get_mcp_tools(self):
        """
            加载MCP-Server对应的工具列表
        """
        try:
            if not self.tools:
                # ...原有工具加载逻辑...
                tools, cleanup = await convert_mcp_to_langchain_tools(
                    self.mcp_configs,
                    server_logger
                )
                self.tools = tools
                self.cleanup = cleanup
            # 打印工具列表
            self.server_logger_tools_info()
            server_logger.info(f"tools init success")
        except Exception as e:
            handler_err(server_logger, e, err_name="工具加载失败")
        return self.tools
    

    def server_logger_tools_info(self):
        """
            打印 MCP 模型工具列表
        """
        server_logger.info(f"MCP 模型工具总共[{len(self.tools)}]个")
        server_logger.info(f"MCP 模型工具列表分别如下：")
        for tool in self.tools:
            server_logger.info("=" * 50)
            server_logger.info(f"工具名称: {tool.name}")
            server_logger.info(f"工具描述: {tool.description}")
            server_logger.info(f"工具参数: {tool.args}")

    async def close(self):
        """
            关闭MCP服务端连接资源
        """
        if self.cleanup is not None:
            try:
                await self.cleanup()
                server_logger.info("✅ MCP 资源已成功清理")
            except Exception as e:
                server_logger.error(f"⚠️ 清理 MCP 资源失败: {e}")