wangzhibo
2026-08-31 17ac2ce0ed52874616c79c910667e41e34604826
xiaozhi-esp32-server/main/xiaozhi-server/core/providers/tools/server_mcp/mcp_manager.py
@@ -2,6 +2,7 @@
import asyncio
import os
import sys
import json
from typing import Dict, Any, List
@@ -33,18 +34,33 @@
    def load_config(self) -> Dict[str, Any]:
        """加载MCP服务配置"""
        if len(self.config_path) == 0:
            return {}
        servers: Dict[str, Any] = {}
        if self.config_path:
            try:
                with open(self.config_path, "r", encoding="utf-8") as f:
                    config = json.load(f)
                servers = config.get("mcpServers", {}) or {}
            except Exception as e:
                logger.bind(tag=TAG).error(
                    f"Error loading MCP config from {self.config_path}: {e}"
                )
                servers = {}
        return self._ensure_builtin_servers(servers)
        try:
            with open(self.config_path, "r", encoding="utf-8") as f:
                config = json.load(f)
            return config.get("mcpServers", {})
        except Exception as e:
            logger.bind(tag=TAG).error(
                f"Error loading MCP config from {self.config_path}: {e}"
            )
            return {}
    def _ensure_builtin_servers(self, servers: Dict[str, Any]) -> Dict[str, Any]:
        """Register the Phase 1 translate MCP even if data/.mcp_server_settings.json is missing."""
        if "english-translate" in servers:
            return servers
        script = os.path.join(
            get_project_dir(), "mcp_servers", "translate", "server.py"
        )
        if os.path.isfile(script):
            servers["english-translate"] = {
                "command": sys.executable or "python",
                "args": [script],
            }
            logger.bind(tag=TAG).info("已注册内置 english-translate MCP")
        return servers
    async def _init_server(self, name: str, srv_config: Dict[str, Any]):
        """初始化单个MCP服务"""