path: root/mcp/install.py

#!/usr/bin/env python3
"""Install / uninstall / check MCP servers at user scope.

Three modes:

  install (default)
    Reads structure from mcp/servers.json (placeholders ${VAR} for secrets),
    decrypts mcp/secrets.env.gpg via gpg-agent (pinentry will prompt; cached
    per gpg-agent.conf), expands placeholders, registers anything not already
    in `claude mcp list`. Idempotent — re-running is safe.

  --uninstall
    Iterates over servers.json and runs `claude mcp remove <name> -s user`
    for each. "Not registered" errors are ignored. Idempotent.

  --check
    Dry-run drift report. Lists each server in servers.json as MISSING
    (in config, not registered) or ok (in both); lists each registered
    server not in servers.json as EXTRA. No secrets decryption, no writes.
"""
from __future__ import annotations

import argparse
import base64
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Any

REPO = Path(__file__).resolve().parent.parent
SERVERS = REPO / "mcp" / "servers.json"
SECRETS_GPG = REPO / "mcp" / "secrets.env.gpg"
GCP_KEYS_OUT = REPO / "mcp" / "gcp-oauth.keys.json"
GOOGLE_DOCS_TOKENS_DIR = Path.home() / ".config" / "google-docs-mcp"
GOOGLE_DOCS_PROFILES = ("personal", "work")


def expand(value: Any, env: dict[str, str]) -> Any:
    pattern = re.compile(r"\$\{(\w+)\}")
    if isinstance(value, str):
        def repl(m: re.Match) -> str:
            key = m.group(1)
            if key in env:
                return env[key]
            if key in os.environ:
                return os.environ[key]
            raise SystemExit(f"ERROR: ${{{key}}} not defined in secrets.env or process env")
        return pattern.sub(repl, value)
    if isinstance(value, list):
        return [expand(v, env) for v in value]
    if isinstance(value, dict):
        return {k: expand(v, env) for k, v in value.items()}
    return value


def decrypt_secrets() -> dict[str, str]:
    if not SECRETS_GPG.is_file():
        raise SystemExit(f"ERROR: {SECRETS_GPG} missing")
    print(f"Decrypting {SECRETS_GPG.relative_to(REPO)}...")
    res = subprocess.run(
        ["gpg", "-d", "--quiet", str(SECRETS_GPG)],
        capture_output=True, text=True
    )
    if res.returncode != 0:
        raise SystemExit(f"ERROR: gpg decryption failed:\n{res.stderr}")
    env: dict[str, str] = {}
    for raw in res.stdout.splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        key, _, val = line.partition("=")
        env[key.strip()] = val.strip()
    return env


def already_registered() -> set[str]:
    res = subprocess.run(["claude", "mcp", "list"], capture_output=True, text=True)
    names: set[str] = set()
    for line in res.stdout.splitlines():
        if ":" not in line:
            continue
        head = line.split(":", 1)[0].strip()
        if head and head[0].isalnum():
            names.add(head)
    return names


def build_add_cmd(name: str, cfg: dict[str, Any]) -> list[str] | None:
    base = ["claude", "mcp", "add", "--scope", "user"]
    transport = cfg.get("type")
    if transport in ("http", "sse"):
        return base + ["--transport", transport, name, cfg["url"]]
    if transport == "stdio":
        # commander.js parses -e as variadic; placing it before the positional
        # name lets the parser eat the name as another -e value. Putting -e
        # after the name (and before --) avoids that.
        env_flags: list[str] = []
        for k, v in cfg.get("env", {}).items():
            env_flags += ["-e", f"{k}={v}"]
        return base + [name] + env_flags + ["--", cfg["command"], *cfg.get("args", [])]
    return None


def check_mode() -> int:
    """Drift report: compare servers.json against claude mcp list."""
    if subprocess.run(["which", "claude"], capture_output=True).returncode != 0:
        print("ERROR: claude not found in PATH", file=sys.stderr)
        return 1
    if not SERVERS.is_file():
        print(f"ERROR: {SERVERS} missing", file=sys.stderr)
        return 1

    configured = set(json.loads(SERVERS.read_text()).keys())
    registered = already_registered()

    print("MCP server drift (servers.json vs `claude mcp list`):")
    missing = sorted(configured - registered)
    extra = sorted(registered - configured)
    in_both = sorted(configured & registered)

    for name in in_both:
        print(f"  ok       {name}")
    for name in missing:
        print(f"  MISSING  {name} (in servers.json, not registered)")
    for name in extra:
        print(f"  EXTRA    {name} (registered, not in servers.json)")

    print(
        f"\nSummary: {len(in_both)} ok, {len(missing)} missing, {len(extra)} extra"
    )
    # Exit non-zero only on MISSING (since `make install-mcp` would fix it).
    # EXTRA entries are often intentional manual additions (web-app servers,
    # one-off experiments) and shouldn't fail the check.
    return 1 if missing else 0


def uninstall_mode() -> int:
    """Remove every server listed in servers.json from `claude mcp list`."""
    if subprocess.run(["which", "claude"], capture_output=True).returncode != 0:
        print("ERROR: claude not found in PATH", file=sys.stderr)
        return 1
    if not SERVERS.is_file():
        print(f"ERROR: {SERVERS} missing", file=sys.stderr)
        return 1

    configured = json.loads(SERVERS.read_text()).keys()
    registered = already_registered()

    print("Uninstalling MCP servers (user scope):")
    removed = skipped = 0
    for name in configured:
        if name not in registered:
            print(f"  skip   {name} (not registered)")
            skipped += 1
            continue
        res = subprocess.run(
            ["claude", "mcp", "remove", name, "-s", "user"],
            capture_output=True, text=True
        )
        if res.returncode == 0:
            print(f"  rm     {name}")
            removed += 1
        else:
            err = (res.stderr or res.stdout).strip().splitlines()[-1][:120]
            print(f"  FAIL   {name}: {err}")

    print(f"\nDone. Removed: {removed}  Skipped: {skipped}")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Install, uninstall, or check MCP servers (user scope)."
    )
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument(
        "--check", action="store_true",
        help="Dry-run drift report (no decryption, no writes)."
    )
    mode.add_argument(
        "--uninstall", action="store_true",
        help="Remove every server listed in servers.json from claude mcp list."
    )
    args = parser.parse_args()

    if args.check:
        return check_mode()
    if args.uninstall:
        return uninstall_mode()

    for tool in ("gpg", "claude"):
        if subprocess.run(["which", tool], capture_output=True).returncode != 0:
            print(f"ERROR: {tool} not found in PATH", file=sys.stderr)
            return 1

    if not SERVERS.is_file():
        print(f"ERROR: {SERVERS} missing", file=sys.stderr)
        return 1

    env = decrypt_secrets()

    # Extract bundled OAuth keys JSON, write to disk at the path
    # google-calendar-mcp will read at runtime
    if "GCP_OAUTH_KEYS_JSON_B64" in env:
        GCP_KEYS_OUT.write_bytes(base64.b64decode(env.pop("GCP_OAUTH_KEYS_JSON_B64")))
        GCP_KEYS_OUT.chmod(0o600)
        env["GOOGLE_OAUTH_CREDENTIALS_PATH"] = str(GCP_KEYS_OUT.resolve())
        print(f"  wrote  {GCP_KEYS_OUT.relative_to(REPO)} (mode 600)")

    # Extract bundled @a-bonus/google-docs-mcp OAuth tokens, write each to the
    # per-profile path the package reads at startup. Without this, a fresh
    # machine has no token cache and the server falls back to interactive
    # OAuth — which Claude Code's stdio MCP loader can't drive, so it shows
    # "Failed to connect" until the user runs the npx command manually.
    for profile in GOOGLE_DOCS_PROFILES:
        var = f"GOOGLE_DOCS_{profile.upper()}_TOKEN_B64"
        if var not in env:
            continue
        target_dir = GOOGLE_DOCS_TOKENS_DIR / profile
        target_dir.mkdir(parents=True, exist_ok=True)
        target_dir.chmod(0o700)
        target = target_dir / "token.json"
        target.write_bytes(base64.b64decode(env.pop(var)))
        target.chmod(0o600)
        print(f"  wrote  {target} (mode 600)")

    existing = already_registered()
    servers = json.loads(SERVERS.read_text())

    print("\nRegistering MCP servers (user scope):")
    added = skipped = failed = 0
    for name, raw_cfg in servers.items():
        if name in existing:
            print(f"  skip   {name} (already registered)")
            skipped += 1
            continue

        cfg = expand(raw_cfg, env)
        cmd = build_add_cmd(name, cfg)
        if cmd is None:
            print(f"  FAIL   {name} (unknown transport: {cfg.get('type')!r})")
            failed += 1
            continue

        res = subprocess.run(cmd, capture_output=True, text=True)
        if res.returncode == 0:
            transport = cfg["type"]
            tail = cfg.get("url") or f"{cfg.get('command', '')} {' '.join(cfg.get('args', []))[:60]}"
            print(f"  add    {name} ({transport}: {tail})")
            added += 1
        else:
            err = (res.stderr or res.stdout).strip().splitlines()[-1][:120]
            print(f"  FAIL   {name}: {err}")
            failed += 1

    print(f"\nDone. Added: {added}  Skipped: {skipped}  Failed: {failed}")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())