Spaces:
Sleeping
Sleeping
File size: 4,589 Bytes
f90aaa2 | 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """MCP Server Manager for Open-LLM-Vtuber."""
import shutil
import json
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
DEFAULT_CONFIG_PATH = "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."""
path_obj = Path(config_path)
if not path_obj.exists():
default_data = {
"mcp_servers": {
"time": {
"command": "uvx",
"args": ["mcp-server-time", "--local-timezone=Asia/Makassar"]
},
"ddg-search": {
"command": "uvx",
"args": ["duckduckgo-mcp-server"]
}
}
}
try:
path_obj.write_text(json.dumps(default_data, indent=4), encoding="utf-8")
logger.info(f"MCPSR: Created default '{config_path}' configuration.")
except Exception as e:
logger.error(f"MCPSR: Failed to create default '{config_path}': {e}")
try:
config_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(
config_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 config file."""
servers_config: Dict[str, Dict[str, Any]] = self.config.get("mcp_servers", {})
if servers_config == {}:
logger.warning("MCPSR: No servers found in the config file.")
return
for server_name, server_details in servers_config.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":
if not self.npx_available:
logger.warning(
f"MCPSR: npx is not available. Cannot load server '{server_name}'."
)
continue
elif command == "uvx":
if not self.uvx_available:
logger.warning(
f"MCPSR: uvx is not available. Cannot load server '{server_name}'."
)
continue
elif command == "node":
if not self.node_available:
logger.warning(
f"MCPSR: node is not available. Cannot load server '{server_name}'."
)
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)
|