#!/usr/bin/env python3 """Small MCP client for MCPHub's official management API. This server deliberately exposes the common installation path (remote HTTP, npx, uvx) without giving an agent a shell on Unraid. MCPHub remains the single source of truth and performs process supervision itself. """ from __future__ import annotations import json import os import pathlib import re import urllib.error import urllib.parse import urllib.request from typing import Any from mcp.server.fastmcp import FastMCP API_BASE = os.environ.get("MCPHUB_API_URL", "http://127.0.0.1:3000/api").rstrip("/") TOKEN_FILE = pathlib.Path(os.environ.get("MCPHUB_API_TOKEN_FILE", "/app/data/client-token")) CLIENT_GROUP = os.environ.get("MCPHUB_CLIENT_GROUP", "hermes").strip() NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,62}$") NPM_PACKAGE_RE = re.compile(r"^(?:@[A-Za-z0-9._-]+/)?[A-Za-z0-9._-]+(?:@[A-Za-z0-9._+~-]+)?$") PYTHON_PACKAGE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*(?:==[A-Za-z0-9._+~-]+)?$") MAX_SERVERS = 80 MAX_TOOLS = 80 mcp = FastMCP( "mcphub-admin", instructions=( "Manage MCP servers through MCPHub's official API. Prefer HTTP MCPs or " "pinned npx/uvx packages. Inspect first, install once, then verify the " "connection and tools. Removal requires the user's explicit request." ), ) class HubError(RuntimeError): pass def _token() -> str: try: value = TOKEN_FILE.read_text(encoding="utf-8").strip() except OSError as exc: raise HubError("MCPHub API token is unavailable") from exc if not value: raise HubError("MCPHub API token is empty") return value def _request(method: str, path: str, body: dict[str, Any] | None = None) -> Any: payload = None if body is None else json.dumps(body).encode("utf-8") request = urllib.request.Request( API_BASE + path, data=payload, method=method, headers={ "Authorization": f"Bearer {_token()}", "Accept": "application/json", "Content-Type": "application/json", }, ) try: with urllib.request.urlopen(request, timeout=30) as response: raw = response.read(2_000_000) except urllib.error.HTTPError as exc: detail = exc.read(800).decode("utf-8", errors="replace").replace("\n", " ") raise HubError(f"MCPHub API returned HTTP {exc.code}: {detail[:500]}") from exc except OSError as exc: raise HubError(f"MCPHub API is unreachable: {str(exc)[:300]}") from exc try: return json.loads(raw) except json.JSONDecodeError as exc: raise HubError("MCPHub API returned invalid JSON") from exc def _name(value: str) -> str: value = value.strip() if not NAME_RE.fullmatch(value): raise HubError("Name must contain only letters, numbers, dot, underscore or hyphen") return value def _package(value: str, pattern: re.Pattern[str]) -> str: value = value.strip() if not pattern.fullmatch(value): raise HubError("Invalid package name or version") return value def _args(value: list[str] | None) -> list[str]: result = [str(item) for item in (value or [])] if len(result) > 20 or any(len(item) > 300 for item in result): raise HubError("At most 20 bounded arguments are allowed") return result SECRET_KEYS = re.compile(r"(?i)(authorization|password|passwd|secret|token|api.?key|cookie)") def _redact(value: Any) -> Any: if isinstance(value, dict): return { str(key): ("[configured]" if SECRET_KEYS.search(str(key)) else _redact(item)) for key, item in value.items() } if isinstance(value, list): return [_redact(item) for item in value[:MAX_TOOLS]] if isinstance(value, str) and len(value) > 500: return value[:500] + "..." return value def _compact_server(item: dict[str, Any]) -> dict[str, Any]: tools = item.get("tools") if isinstance(item.get("tools"), list) else [] config = item.get("config") if isinstance(item.get("config"), dict) else {} return { "name": item.get("name"), "status": item.get("status"), "enabled": config.get("enabled", True), "type": config.get("type"), "tool_count": len(tools), "tools": [tool.get("name") for tool in tools[:MAX_TOOLS] if isinstance(tool, dict)], } def _install(name: str, config: dict[str, Any]) -> str: name = _name(name) existing = _request("GET", "/servers") rows = existing.get("data", []) if isinstance(existing, dict) else [] if any(isinstance(row, dict) and row.get("name") == name for row in rows): raise HubError(f"Server already exists: {name}; inspect or update it instead") result = _request("POST", "/servers", {"name": name, "config": config}) group_result = _add_to_group(name) verified = _request("GET", f"/servers/{name}") data = verified.get("data", verified) if isinstance(verified, dict) else verified return json.dumps({ "installed": True, "published_to": CLIENT_GROUP, "server": _redact(data), "api_result": _redact(result), "group_result": _redact(group_result), }, ensure_ascii=False) def _group_id(name: str) -> str: result = _request("GET", "/groups") rows = result.get("data", []) if isinstance(result, dict) else [] match = next( (row for row in rows if isinstance(row, dict) and (row.get("name") == name or row.get("id") == name)), None, ) if not isinstance(match, dict) or not match.get("id"): raise HubError(f"MCPHub group does not exist: {name}") return str(match["id"]) def _add_to_group(name: str, tools: list[str] | None = None) -> Any: if not CLIENT_GROUP: return {"skipped": "no-client-group"} group = urllib.parse.quote(_group_id(CLIENT_GROUP), safe="") configs = _request("GET", f"/groups/{group}/server-configs") rows = configs.get("data", []) if isinstance(configs, dict) else [] present = any(isinstance(row, dict) and row.get("name") == name for row in rows) result: Any = {"already_present": True} if not present: result = _request("POST", f"/groups/{group}/servers", {"serverName": name}) if tools is not None: result = _request( "PUT", f"/groups/{group}/server-configs/{urllib.parse.quote(name, safe='')}/tools", {"tools": tools}, ) return result @mcp.tool() def mcphub_admin_list_servers() -> str: """List configured MCPHub servers with connection state and compact tool names.""" result = _request("GET", "/servers") rows = result.get("data", []) if isinstance(result, dict) else [] compact = [_compact_server(row) for row in rows[:MAX_SERVERS] if isinstance(row, dict)] return json.dumps({"count": len(rows), "servers": compact, "truncated": len(rows) > MAX_SERVERS}, ensure_ascii=False) @mcp.tool() def mcphub_admin_get_server(name: str) -> str: """Inspect one MCPHub server, its status, transport and exposed tools. Secrets are redacted.""" name = _name(name) result = _request("GET", f"/servers/{name}") return json.dumps(_redact(result), ensure_ascii=False) @mcp.tool() def mcphub_admin_install_http(name: str, url: str, enabled: bool = True) -> str: """Install a remote Streamable-HTTP MCP by URL and verify registration. Use OAuth-capable entries through the UI when interactive login is required.""" url = url.strip() if not re.match(r"^https?://", url): raise HubError("HTTP MCP URL must start with http:// or https://") return _install(name, {"type": "streamable-http", "url": url, "enabled": bool(enabled), "owner": "admin"}) @mcp.tool() def mcphub_admin_install_npx(name: str, package: str, arguments: list[str] | None = None, enabled: bool = True) -> str: """Install an npm-distributed stdio MCP with npx. Pin package@version whenever possible; arguments are passed without a shell.""" package = _package(package, NPM_PACKAGE_RE) args = ["-y", package, *_args(arguments)] return _install(name, {"type": "stdio", "command": "npx", "args": args, "enabled": bool(enabled), "owner": "admin"}) @mcp.tool() def mcphub_admin_install_uvx(name: str, package: str, arguments: list[str] | None = None, enabled: bool = True) -> str: """Install a Python-distributed stdio MCP with uvx. Pin package==version whenever possible; arguments are passed without a shell.""" package = _package(package, PYTHON_PACKAGE_RE) return _install(name, {"type": "stdio", "command": "uvx", "args": [package, *_args(arguments)], "enabled": bool(enabled), "owner": "admin"}) @mcp.tool() def mcphub_admin_set_enabled(name: str, enabled: bool) -> str: """Enable or disable one explicitly named MCPHub server.""" name = _name(name) result = _request("POST", f"/servers/{name}/toggle", {"enabled": bool(enabled)}) return json.dumps(_redact(result), ensure_ascii=False) @mcp.tool() def mcphub_admin_reload(name: str) -> str: """Reconnect or respawn one explicitly named MCPHub server after a change.""" name = _name(name) result = _request("POST", f"/servers/{name}/reload") return json.dumps(_redact(result), ensure_ascii=False) @mcp.tool() def mcphub_admin_publish_to_hermes(name: str, tools: list[str] | None = None) -> str: """Publish one installed server to the central Hermes group. Optionally expose only the named tools; omit tools to expose all.""" name = _name(name) selected = None if tools is None else _args(tools) result = _add_to_group(name, selected) return json.dumps({"published": name, "group": CLIENT_GROUP, "result": _redact(result)}, ensure_ascii=False) @mcp.tool() def mcphub_admin_unpublish_from_hermes(name: str) -> str: """Remove one server from the central Hermes group without uninstalling or disabling the server itself.""" name = _name(name) group = urllib.parse.quote(_group_id(CLIENT_GROUP), safe="") result = _request("DELETE", f"/groups/{group}/servers/{urllib.parse.quote(name, safe='')}") return json.dumps({"unpublished": name, "group": CLIENT_GROUP, "result": _redact(result)}, ensure_ascii=False) @mcp.tool() def mcphub_admin_remove(name: str, confirmation: str) -> str: """Permanently remove a server only after the user explicitly requested it. confirmation must exactly equal REMOVE:.""" name = _name(name) if confirmation != f"REMOVE:{name}": raise HubError(f"Confirmation must exactly equal REMOVE:{name}") result = _request("DELETE", f"/servers/{name}") return json.dumps(_redact(result), ensure_ascii=False) if __name__ == "__main__": mcp.run(transport="stdio")