| | |
| | | |
| | | import asyncio |
| | | import os |
| | | import sys |
| | | import json |
| | | from typing import Dict, Any, List |
| | | |
| | |
| | | |
| | | 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服务""" |