AI-Profile-Router/platform/mcphub/configure-settings.py

#!/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 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
    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()