let Hermes manage MCPHub through its API

2026-08-26 11:28:51 +02:00
parent bc902261b9
commit 2595bab17b
11 changed files with 377 additions and 34 deletions
@@ -0,0 +1,211 @@
+#!/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.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"))
+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})
+    verified = _request("GET", f"/servers/{name}")
+    data = verified.get("data", verified) if isinstance(verified, dict) else verified
+    return json.dumps({"installed": True, "server": _redact(data), "api_result": _redact(result)}, ensure_ascii=False)
+
+
+@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_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(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")