1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
#!/usr/bin/env python3
"""Install MCP servers at user scope.
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, then registers anything not already in `claude mcp list`.
Idempotent — re-running is safe.
"""
from __future__ import annotations
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 main() -> int:
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())
|
