198 lines
7.0 KiB
Python
198 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Replace demo MCPHub entries with CasaDeRoll's declarative server set.
|
|
|
|
The existing users, bearer keys, prompts and resources are preserved. Secret
|
|
values are read locally and written only to the runtime settings file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import secrets
|
|
import tempfile
|
|
import uuid
|
|
|
|
|
|
def env_file(path: pathlib.Path) -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
if not path.is_file():
|
|
return values
|
|
for raw in path.read_text(encoding="utf-8").splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
key, value = key.strip(), value.strip()
|
|
if key.startswith("export "):
|
|
key = key[7:].strip()
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
|
|
value = value[1:-1]
|
|
values[key] = value
|
|
return values
|
|
|
|
|
|
PLACEHOLDER = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")
|
|
|
|
|
|
def expand(value: object, values: dict[str, str], source: pathlib.Path) -> object:
|
|
"""Resolve secret placeholders without ever logging their values."""
|
|
if isinstance(value, str):
|
|
def replace(match: re.Match[str]) -> str:
|
|
key = match.group(1)
|
|
resolved = values.get(key, "").strip()
|
|
if not resolved:
|
|
raise SystemExit(f"{key} is missing in {source}")
|
|
return resolved
|
|
return PLACEHOLDER.sub(replace, value)
|
|
if isinstance(value, list):
|
|
return [expand(item, values, source) for item in value]
|
|
if isinstance(value, dict):
|
|
return {key: expand(item, values, source) for key, item in value.items()}
|
|
return value
|
|
|
|
|
|
def registry_servers(registry: pathlib.Path, secrets_dir: pathlib.Path,
|
|
existing: dict[str, object]) -> dict[str, object]:
|
|
document = json.loads(registry.read_text(encoding="utf-8"))
|
|
if document.get("version") != 1:
|
|
raise SystemExit("Unsupported MCP registry schema")
|
|
result: dict[str, object] = {}
|
|
for item in document.get("servers", []):
|
|
spec = item.get("hub")
|
|
if not isinstance(spec, dict):
|
|
continue
|
|
server_id = str(item.get("hermes_id") or item["id"])
|
|
spec = dict(spec)
|
|
secret_name = str(spec.pop("secret_file", ""))
|
|
secret_path = secrets_dir / secret_name if secret_name else secrets_dir
|
|
values = env_file(secret_path) if secret_name else {}
|
|
rendered = expand(spec, values, secret_path)
|
|
if isinstance(rendered, dict) and isinstance(rendered.get("url"), str):
|
|
rendered["url"] = re.sub(r"(?<!:)//+", "/", rendered["url"])
|
|
previous = existing.get(server_id)
|
|
deployment = item.get("deployment") or {}
|
|
if (not deployment.get("managed_enabled")
|
|
and isinstance(previous, dict) and "enabled" in previous):
|
|
rendered["enabled"] = bool(previous["enabled"])
|
|
result[server_id] = rendered
|
|
return result
|
|
|
|
|
|
def ensure_hermes_group(settings: dict[str, object]) -> None:
|
|
"""Seed the bounded Hermes aggregate route on fresh installs/recovery."""
|
|
server_names = [
|
|
"mcphub-admin", "athena-operator", "github", "arr", "navidrome",
|
|
"unraid",
|
|
]
|
|
members: list[dict[str, object]] = [
|
|
{"name": name, "tools": "all", "prompts": "all", "resources": "all"}
|
|
for name in server_names
|
|
]
|
|
members.append({
|
|
"name": "fritzbox",
|
|
"tools": ["list_services", "list_actions", "describe_action", "call_action"],
|
|
"prompts": [],
|
|
"resources": [],
|
|
})
|
|
groups = settings.setdefault("groups", [])
|
|
if not isinstance(groups, list):
|
|
groups = []
|
|
settings["groups"] = groups
|
|
current = next(
|
|
(group for group in groups
|
|
if isinstance(group, dict) and group.get("name") == "hermes"),
|
|
None,
|
|
)
|
|
if current is None:
|
|
current = {
|
|
"id": str(uuid.uuid4()),
|
|
"owner": "system",
|
|
"name": "hermes",
|
|
}
|
|
groups.append(current)
|
|
current["description"] = "Gefilterte zentrale MCP-Auswahl für alle Hermes-Clients"
|
|
current["servers"] = members
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("settings", type=pathlib.Path)
|
|
parser.add_argument("secrets", type=pathlib.Path)
|
|
parser.add_argument(
|
|
"--registry", type=pathlib.Path,
|
|
default=pathlib.Path("/opt/casaderoll/config/mcp-registry.json"),
|
|
)
|
|
parser.add_argument("--web-backend", default="", help="TinySearch MCP URL; empty keeps web disabled")
|
|
parser.add_argument("--searxng", default="", help="SearXNG base URL")
|
|
args = parser.parse_args()
|
|
|
|
args.settings.parent.mkdir(parents=True, exist_ok=True)
|
|
settings = json.loads(args.settings.read_text(encoding="utf-8")) if args.settings.exists() else {}
|
|
servers = registry_servers(
|
|
args.registry,
|
|
args.secrets,
|
|
settings.get("mcpServers", {}) if isinstance(settings.get("mcpServers"), dict) else {},
|
|
)
|
|
if args.web_backend:
|
|
servers["web"] = {
|
|
"type": "stdio",
|
|
"command": "python3",
|
|
"args": ["/opt/casaderoll/mcps/web_search_mcp.py"],
|
|
"env": {
|
|
"TINYSEARCH_MCP_URL": args.web_backend,
|
|
"SEARXNG_URL": args.searxng,
|
|
},
|
|
"enabled": True,
|
|
}
|
|
|
|
settings["mcpServers"] = servers
|
|
ensure_hermes_group(settings)
|
|
settings.setdefault("users", [])
|
|
token_path = args.settings.parent / "client-token"
|
|
keys = settings.setdefault("bearerKeys", [])
|
|
client_key = next((item for item in keys if item.get("name") == "casaderoll-clients"), None)
|
|
if client_key is None:
|
|
token = secrets.token_urlsafe(48)
|
|
client_key = {
|
|
"id": str(uuid.uuid4()),
|
|
"name": "casaderoll-clients",
|
|
"token": token,
|
|
"enabled": True,
|
|
"kind": "system",
|
|
"accessType": "all",
|
|
"allowedGroups": [],
|
|
"allowedServers": [],
|
|
}
|
|
keys.append(client_key)
|
|
else:
|
|
token = str(client_key["token"])
|
|
client_key["enabled"] = True
|
|
client_key["accessType"] = "all"
|
|
token_path.write_text(token + "\n", encoding="utf-8")
|
|
os.chmod(token_path, 0o600)
|
|
settings.setdefault("prompts", [])
|
|
settings.setdefault("resources", [])
|
|
system = settings.setdefault("systemConfig", {})
|
|
system.setdefault("routing", {})["skipAuth"] = False
|
|
|
|
fd, temporary = tempfile.mkstemp(prefix=".mcp-settings-", dir=args.settings.parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
json.dump(settings, handle, indent=2, ensure_ascii=False)
|
|
handle.write("\n")
|
|
os.chmod(temporary, 0o600)
|
|
os.replace(temporary, args.settings)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
print("MCPHUB_SETTINGS_CONFIGURED")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|