| """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
|
|
|
|
|
| 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."""
|
|
|
| self.config_path = config_path
|
|
|
| try:
|
|
|
| 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."""
|
|
|
|
|
| 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) |
