""" mcp_core.py -- Pure logic for the Claude MCP config manager. No GUI imports live in here on purpose: every function below is unit-testable and could just as easily back a CLI. The GUI (mcp_manager.py) is a thin shell over these functions. The cardinal rule of this module: when writing a config back to disk we ONLY ever touch the `mcpServers` block (and our own `_disabledMcpServers` parking key). Every other top-level key in the user's config is preserved verbatim and in its original position. """ from __future__ import annotations import functools import glob import json import os import shutil import sys import tempfile import time from dataclasses import dataclass, field from pathlib import Path from urllib.parse import urlparse CONFIG_FILENAME = "claude_desktop_config.json" # Disabled servers are parked under this non-standard key. Claude Desktop only # reads `mcpServers`, so anything here is ignored by the app but kept on disk so # we can toggle it back on without losing the definition. DISABLED_KEY = "_disabledMcpServers" BACKUP_DIRNAME = ".bcc_backups" MAX_BACKUPS = 15 # Fields the editor knows how to render. Anything else on a server object is # considered "extra" and is preserved untouched on save. KNOWN_FIELDS = {"command", "args", "env", "url", "type", "headers"} # --------------------------------------------------------------------------- # # Data model # --------------------------------------------------------------------------- # @dataclass class Profile: """A discovered (or manually added) Claude install location.""" label: str path: Path config_exists: bool def __post_init__(self): self.path = Path(self.path) @dataclass class ServerEntry: name: str data: dict enabled: bool = True @property def kind(self) -> str: return "remote" if "url" in self.data and "command" not in self.data else "stdio" # --------------------------------------------------------------------------- # # Discovery # --------------------------------------------------------------------------- # def app_support_base() -> Path: """The per-platform directory that holds the `Claude*` data folders.""" if sys.platform == "darwin": return Path.home() / "Library" / "Application Support" if os.name == "nt": return Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming")) return Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) def discover_profiles() -> list[Profile]: """ Find every `Claude*` data directory in the platform's app-support base. This catches the standard `Claude` folder plus any sibling like `Claude-Work`, `Claude-Personal`, etc. A directory is listed even if the config file doesn't exist yet (it will be created on first save). """ base = app_support_base() out: list[Profile] = [] if base.is_dir(): seen = set() for d in sorted(base.glob("Claude*")): if d.is_dir() and d.name not in seen: seen.add(d.name) cfg = d / CONFIG_FILENAME out.append(Profile(label=d.name, path=cfg, config_exists=cfg.is_file())) return out def profile_from_path(path: str | os.PathLike) -> Profile: """Build a Profile from a user-supplied config path (the 'override' case).""" p = Path(path) # Prefer the parent folder name as the label (e.g. .../Claude-Work/...json -> Claude-Work) label = p.parent.name or p.name return Profile(label=label, path=p, config_exists=p.is_file()) # --------------------------------------------------------------------------- # # Load / extract / apply # --------------------------------------------------------------------------- # def load_config(path: str | os.PathLike) -> dict: """Read the full config dict. Missing/empty file -> {}. Bad JSON -> raises.""" p = Path(path) if not p.is_file(): return {} text = p.read_text(encoding="utf-8") if not text.strip(): return {} obj = json.loads(text) # JSONDecodeError bubbles up for the GUI to display if not isinstance(obj, dict): raise ValueError("Top-level JSON in the config file is not an object.") return obj def extract_servers(cfg: dict) -> list[ServerEntry]: """Pull enabled (`mcpServers`) and disabled (`_disabledMcpServers`) servers.""" out: list[ServerEntry] = [] for name, data in (cfg.get("mcpServers") or {}).items(): out.append(ServerEntry(name=name, data=dict(data), enabled=True)) for name, data in (cfg.get(DISABLED_KEY) or {}).items(): out.append(ServerEntry(name=name, data=dict(data), enabled=False)) return out def apply_servers(cfg: dict, servers: list[ServerEntry]) -> dict: """ Write the server list back into `cfg` in place, preserving every other key and the position of `mcpServers`. Returns the same dict for convenience. """ enabled = {s.name: s.data for s in servers if s.enabled} disabled = {s.name: s.data for s in servers if not s.enabled} cfg["mcpServers"] = enabled # replaces value if key existed; appends otherwise if disabled: cfg[DISABLED_KEY] = disabled else: cfg.pop(DISABLED_KEY, None) return cfg # --------------------------------------------------------------------------- # # Write (atomic, with rotating backups) # --------------------------------------------------------------------------- # def _make_backup(path: Path) -> Path: bdir = path.parent / BACKUP_DIRNAME bdir.mkdir(exist_ok=True) stamp = time.strftime("%Y%m%d-%H%M%S") dest = bdir / f"{path.stem}.{stamp}.json" # Avoid clobbering a same-second backup n = 1 while dest.exists(): dest = bdir / f"{path.stem}.{stamp}.{n}.json" n += 1 shutil.copy2(path, dest) # Prune oldest beyond MAX_BACKUPS backups = sorted(bdir.glob(f"{path.stem}.*.json")) for old in backups[:-MAX_BACKUPS]: try: old.unlink() except OSError: pass return dest def write_config(path: str | os.PathLike, cfg: dict) -> Path | None: """ Atomically write `cfg` to `path` (2-space pretty JSON). Backs up any existing file first. Returns the backup path (or None if there was nothing to back up). """ p = Path(path) p.parent.mkdir(parents=True, exist_ok=True) backup = _make_backup(p) if p.is_file() else None payload = json.dumps(cfg, indent=2, ensure_ascii=False) + "\n" fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=".tmp_mcp_", suffix=".json") try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(payload) os.replace(tmp, p) # atomic on the same filesystem finally: if os.path.exists(tmp): os.remove(tmp) return backup # --------------------------------------------------------------------------- # # Paste / import parsing # --------------------------------------------------------------------------- # def looks_like_server(data) -> bool: return isinstance(data, dict) and ("command" in data or "url" in data) def suggest_name(data: dict) -> str: """Derive a friendly default name from a server definition.""" if "url" in data and "command" not in data: host = (urlparse(data["url"]).hostname or "remote") return host.split(".")[0] or "remote" for a in data.get("args", []) or []: if isinstance(a, str) and ("/" in a or a.startswith("@")): base = a.rstrip("/").split("/")[-1] base = base.replace("server-", "").replace("mcp-server-", "").replace("mcp-", "") if base: return base cmd = data.get("command", "server") return Path(str(cmd)).stem or "server" def parse_pasted_json(text: str) -> dict[str, dict]: """ Accept any of the shapes MCP docs hand out and return {name: server_dict}: 1. Full config: {"mcpServers": {"name": {...}}} 2. Inner block only: {"name": {"command": ...}} 3. A bare server obj: {"command": ..., "args": [...]} (name is suggested) Raises ValueError with a human message on anything unrecognizable. """ text = (text or "").strip() if not text: raise ValueError("Nothing to parse.") obj = json.loads(text) if not isinstance(obj, dict): raise ValueError("Top-level JSON must be an object ({ ... }).") if isinstance(obj.get("mcpServers"), dict): servers = obj["mcpServers"] elif looks_like_server(obj): servers = {suggest_name(obj): obj} elif obj and all(isinstance(v, dict) for v in obj.values()): servers = obj else: raise ValueError("Couldn't find any MCP server definitions in that JSON.") clean: dict[str, dict] = {} for name, data in servers.items(): if not looks_like_server(data): raise ValueError( f"'{name}' doesn't look like an MCP server " f"(it needs a 'command' or a 'url')." ) clean[str(name)] = data return clean # --------------------------------------------------------------------------- # # Validation # --------------------------------------------------------------------------- # def validate_servers(servers: list[ServerEntry]) -> list[str]: """Return a list of human-readable problems. Empty list == all good.""" problems: list[str] = [] seen: dict[str, int] = {} for s in servers: nm = s.name.strip() if not nm: problems.append("A server has an empty name.") seen[nm] = seen.get(nm, 0) + 1 if s.kind == "stdio": if not str(s.data.get("command", "")).strip(): problems.append(f"'{nm or '(unnamed)'}' is a local server but has no command.") else: url = str(s.data.get("url", "")).strip() if not url: problems.append(f"'{nm or '(unnamed)'}' is a remote server but has no URL.") elif not (url.startswith("http://") or url.startswith("https://")): problems.append(f"'{nm}' has a URL that isn't http(s).") for nm, count in seen.items(): if nm and count > 1: problems.append(f"Duplicate server name: '{nm}' ({count}x).") return problems # --------------------------------------------------------------------------- # # Dependency / PATH checking # --------------------------------------------------------------------------- # @functools.lru_cache(maxsize=1) def augmented_path() -> str: """ PATH + the usual runtime install locations. GUI apps (especially bundled .app launches) often don't inherit the shell PATH, so npx/uvx can appear 'missing' when they're actually fine. This widens the net. Memoized for the session (PATH rarely changes mid-run). Call refresh_path_cache() after the user installs something to re-scan. """ parts = os.environ.get("PATH", "").split(os.pathsep) home = Path.home() extra = [ "/usr/local/bin", "/opt/homebrew/bin", "/usr/bin", "/bin", str(home / ".local" / "bin"), str(home / ".cargo" / "bin"), str(home / ".deno" / "bin"), str(home / ".bun" / "bin"), str(home / ".volta" / "bin"), ] extra += glob.glob(str(home / ".nvm" / "versions" / "node" / "*" / "bin")) extra += glob.glob(str(home / ".local" / "share" / "fnm" / "node-versions" / "*" / "installation" / "bin")) if os.name == "nt": appdata = os.environ.get("APPDATA", "") if appdata: extra.append(str(Path(appdata) / "npm")) seen, ordered = set(), [] for p in parts + extra: if p and p not in seen and Path(p).is_dir(): seen.add(p) ordered.append(p) return os.pathsep.join(ordered) def refresh_path_cache(): """Forget the memoized PATH so the next check re-scans (e.g. after an install).""" augmented_path.cache_clear() # Tailored install advice keyed by the runner's basename. RUNNER_HINTS = { "npx": "Node.js / npx not found. Install Node from nodejs.org or `brew install node`. " "With nvm, the binary lives at ~/.nvm/versions/node//bin.", "node": "Node.js not found. Install from nodejs.org or `brew install node`.", "uvx": "uv not found. Install with `curl -LsSf https://astral.sh/uv/install.sh | sh` " "(or `brew install uv`). uvx ships inside uv.", "uv": "uv not found. Install with `curl -LsSf https://astral.sh/uv/install.sh | sh` " "(or `brew install uv`).", "python": "Python not found on this PATH.", "python3": "Python 3 not found on this PATH.", "docker": "Docker not found. Install Docker Desktop and make sure it's running.", "bun": "Bun not found. Install with `curl -fsSL https://bun.sh/install | bash`.", "deno": "Deno not found. Install with `curl -fsSL https://deno.land/install.sh | sh`.", } def runner_hint(cmd: str) -> str: base = Path(cmd).name.lower() if base.endswith(".exe"): base = base[:-4] if base in RUNNER_HINTS: return RUNNER_HINTS[base] if ("/" in cmd) or ("\\" in cmd): return (f"'{cmd}' looks like a file path, but it doesn't exist or isn't executable. " f"Check the path and that it's marked executable (chmod +x).") return f"'{cmd}' was not found on PATH. Install it, or put the full path to the executable in 'command'." SHELL_WRAPPERS = {"cmd", "cmd.exe", "sh", "bash", "zsh", "powershell", "powershell.exe", "pwsh"} def _commands_to_check(data: dict) -> list[str]: cmd = data.get("command") if not cmd: return [] cmds = [str(cmd)] base = Path(str(cmd)).name.lower() if base in SHELL_WRAPPERS: for a in data.get("args", []) or []: if isinstance(a, str) and not a.startswith(("/", "-")) and " " not in a: cmds.append(a) break return cmds def check_dependency(data: dict, path: str | None = None) -> dict: """ Decide whether a server's command can actually be found, and gather enough context to troubleshoot when it can't. status is one of: 'ok' - resolved on the normal (inherited) PATH; will work anywhere. 'warn' - resolved, but ONLY via an augmented location. Works in a terminal launch; a bundled .app launch of Claude may not see it. 'missing' - not found anywhere. 'remote' - a url-based server (no local command to check). 'unknown' - no command set. Returns: {status, label, resolved, in_base, searched_path, hints, commands, detail} """ if "url" in data and "command" not in data: return {"status": "remote", "label": "remote endpoint", "detail": data.get("url", ""), "resolved": {}, "in_base": {}, "searched_path": "", "hints": [], "commands": []} cmds = _commands_to_check(data) if not cmds: return {"status": "unknown", "label": "no command set", "detail": "", "resolved": {}, "in_base": {}, "searched_path": "", "hints": ["This server has no 'command' to run."], "commands": []} aug = path or augmented_path() resolved: dict[str, str | None] = {} in_base: dict[str, bool] = {} for c in cmds: if (os.sep in c) or ("/" in c): # explicit path, independent of PATH cp = Path(c) ok = cp.exists() and os.access(cp, os.X_OK) resolved[c] = str(cp) if ok else None in_base[c] = ok else: resolved[c] = shutil.which(c, path=aug) in_base[c] = bool(shutil.which(c)) # raw inherited PATH only missing = [c for c, r in resolved.items() if not r] hints: list[str] = [] if missing: for c in missing: hints.append(runner_hint(c)) status = "missing" label = "missing: " + ", ".join(missing) else: nonbase = [c for c in resolved if not in_base[c]] if nonbase: status = "warn" label = "found, but only on a non-standard PATH" for c in nonbase: hints.append( f"'{c}' was found at {resolved[c]}, but only via a location this app added on " f"top of the inherited PATH. If you launch Claude Desktop as an app (not from a " f"terminal), it may not see this. Putting the full path in 'command' is the reliable fix." ) else: status = "ok" label = "found: " + ", ".join(Path(p).name for p in resolved.values()) return {"status": status, "label": label, "resolved": resolved, "in_base": in_base, "searched_path": aug, "hints": hints, "commands": cmds, "detail": ""} def diagnostics_text(name: str, data: dict) -> str: """A copy-pasteable plain-text troubleshooting report for one server.""" res = check_dependency(data) L: list[str] = [] if res["status"] == "remote": L.append(f"Server: {name or '(unnamed)'} [remote]") L.append(f"URL: {data.get('url', '')}") if data.get("type"): L.append(f"Transport: {data['type']}") if data.get("headers"): L.append("Headers: " + ", ".join(data["headers"].keys())) L.append("Status: REMOTE (network reachability is not checked)") return "\n".join(L) L.append(f"Server: {name or '(unnamed)'} [local / stdio]") L.append(f"Command: {data.get('command', '')}") if data.get("args"): L.append("Args: " + " ".join(str(a) for a in data["args"])) if data.get("env"): L.append("Env keys: " + ", ".join(data["env"].keys())) L.append(f"Status: {res['status'].upper()}") L.append("") L.append("Command resolution:") for c, r in res["resolved"].items(): if r: tag = "" if res["in_base"].get(c) else " <-- non-standard PATH; app may not see it" L.append(f" {c} -> {r}{tag}") else: L.append(f" {c} -> NOT FOUND") if res["hints"]: L.append("") L.append("Hints:") for h in res["hints"]: L.append(f" - {h}") base_dirs = [p for p in os.environ.get("PATH", "").split(os.pathsep) if p] aug_dirs = res["searched_path"].split(os.pathsep) if res["searched_path"] else [] extra = [d for d in aug_dirs if d not in base_dirs] L.append("") L.append(f"PATH searched ({len(aug_dirs)} dirs):") for d in aug_dirs: L.append((" + " if d in extra else " ") + d) if extra: L.append("") L.append("(+ = added by this app on top of the PATH it inherited. Claude Desktop, " "launched separately, may NOT have these on its PATH.)") return "\n".join(L) def test_remote(url: str, timeout: float = 5.0) -> tuple[bool, str]: """ Reachability check for a url-based MCP server. ANY HTTP response (even 4xx/5xx) counts as reachable; only connection/DNS/timeout failures count as unreachable. Returns (ok, detail). Network call — run this off the UI thread. """ import socket import urllib.error import urllib.request url = (url or "").strip() if not (url.startswith("http://") or url.startswith("https://")): return False, "not an http(s) URL" req = urllib.request.Request( url, method="GET", headers={"Accept": "text/event-stream, application/json, */*", "User-Agent": "BetterClaudeConfig/1.0"}, ) try: with urllib.request.urlopen(req, timeout=timeout) as r: return True, f"HTTP {r.status}" except urllib.error.HTTPError as e: return True, f"HTTP {e.code} (reachable)" except urllib.error.URLError as e: reason = getattr(e, "reason", e) if isinstance(reason, socket.timeout): return False, "timed out" return False, str(reason) except socket.timeout: return False, "timed out" except Exception as e: # noqa: BLE001 - surface anything else as a clean message return False, str(e) def pin_command_path(data: dict, path: str | None = None) -> tuple[dict, str | None]: """ Fix a 'PATH-risk' (warn) server by rewriting the bare runner name to the absolute path it resolved to, so any process (incl. a bundled Claude.app) can find it. Returns (new_data, note). note is None if nothing was pinned. """ res = check_dependency(data, path) if res["status"] != "warn": return data, None out = dict(data) for c, resolved in res["resolved"].items(): if resolved and not res["in_base"].get(c, True) and ("/" not in c and "\\" not in c): if str(out.get("command", "")) == c: out["command"] = resolved return out, f"command → {resolved}" args = list(out.get("args", []) or []) for i, a in enumerate(args): if a == c: args[i] = resolved out["args"] = args return out, f"'{c}' → {resolved}" return data, None