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