File size: 3,790 Bytes
5669b22 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """MCP Server Manager for Open-LLM-Vtuber."""
import shutil
import json
import os
from pathlib import Path
from typing import Dict, Optional, Union, Any
from loguru import logger
from .types import MCPServer
from .utils.path import validate_file
# Xác định đường dẫn tương đối dựa trên thư mục gốc của dự án
BASE_DIR = Path(__file__).parent.parent.parent.parent
DEFAULT_CONFIG_PATH = Path(os.getcwd()) / "mcp_servers.json"
class ServerRegistry:
"""MCP Server Manager for managing server files."""
def __init__(self, config_path: str | Path = DEFAULT_CONFIG_PATH) -> None:
"""Initialize the MCP Server Manager."""
# PHẢI CÓ DÒNG NÀY: Lưu lại đường dẫn để các hàm khác dùng được
self.config_path = config_path
try:
# Dùng một biến tạm để validate
validated_path = validate_file(config_path, ".json")
except ValueError:
logger.error(
f"MCPSR: File '{config_path}' does not exist, or is not a json file."
)
raise ValueError(
f"MCPSR: File '{config_path}' does not exist, or is not a json file."
)
self.config: Dict[str, Union[str, dict]] = json.loads(
validated_path.read_text(encoding="utf-8")
)
self.servers: Dict[str, MCPServer] = {}
self.npx_available = self._detect_runtime("npx")
self.uvx_available = self._detect_runtime("uvx")
self.node_available = self._detect_runtime("node")
self.load_servers()
def _detect_runtime(self, target: str) -> bool:
"""Check if a runtime is available in the system PATH."""
founded = shutil.which(target)
return True if founded else False
def load_servers(self) -> None:
"""Load servers from the configuration file."""
# Kiểm tra xem self.config có chứa key "mcp_servers" không
# Nếu có, chúng ta lấy nội dung bên trong đó.
servers_data = self.config
if "mcp_servers" in self.config:
self.config = self.config["mcp_servers"]
for server_name, server_details in servers_data.items():
if "command" not in server_details or "args" not in server_details:
logger.warning(
f"MCPSR: Invalid server details for '{server_name}'. Ignoring."
)
continue
command = server_details["command"]
if command == "npx" and not self.npx_available:
continue
elif command == "uvx" and not self.uvx_available:
continue
elif command == "node" and not self.node_available:
continue
self.servers[server_name] = MCPServer(
name=server_name,
command=command,
args=server_details["args"],
env=server_details.get("env", None),
cwd=server_details.get("cwd", None),
timeout=server_details.get("timeout", None),
)
logger.debug(f"MCPSR: Loaded server: '{server_name}'.")
def remove_server(self, server_name: str) -> None:
"""Remove a server from the available servers."""
try:
self.servers.pop(server_name)
logger.info(f"MCPSR: Removed server: {server_name}")
except KeyError:
logger.warning(f"MCPSR: Server '{server_name}' not found. Cannot remove.")
def get_server(self, server_name: str) -> Optional[MCPServer]:
"""Get the server by name."""
return self.servers.get(server_name, None) |