From 5d59c1c42389142c54b5df1f730e826edd94a224 Mon Sep 17 00:00:00 2001 From: the_og Date: Tue, 7 Jul 2026 20:30:03 -0400 Subject: [PATCH] fix: restore actual file content (previous commit had placeholder text by mistake) --- bcc_core.py | 1487 ++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 1486 insertions(+), 1 deletion(-) diff --git a/bcc_core.py b/bcc_core.py index 987bf95..e5b92ff 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -1 +1,1486 @@ -PLACEHOLDER_WILL_REPLACE \ No newline at end of file +""" +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 contextlib +import difflib +import functools +import glob +import json +import os +import re +import shutil +import signal as _signal +import subprocess +import sys +import tempfile +import threading +import time +from dataclasses import dataclass +from pathlib import Path +from typing import NamedTuple +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 + (Claude Desktop installs), then also check for a Claude Code global config. + + Claude Desktop: scans the platform app-support folder for any `Claude*` + directory (catches `Claude`, `Claude-Work`, etc.). + Claude Code: user-scope MCP servers live in ~/.claude.json (that's what + `claude mcp add` writes; project scope is a per-repo .mcp.json, which can + be opened via 'Add config…'). NOT ~/.claude/settings.json — that file is + for permissions/hooks and rejects an mcpServers key with a schema error. + """ + 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())) + + home = Path.home() + cc_cfg = home / ".claude.json" + out.append(Profile(label="Claude Code", path=cc_cfg, config_exists=cc_cfg.is_file())) + + # Legacy: earlier BCC versions (and hand-edits) may have parked servers in + # ~/.claude/settings.json, where Claude Code ignores them. Surface that + # file only when it actually contains an mcpServers block, so the user can + # Copy to ▸ the entries into the real config. + legacy = home / ".claude" / "settings.json" + if legacy.is_file(): + with contextlib.suppress(Exception): + if "mcpServers" in load_config(legacy): + out.append( + Profile( + label="Claude Code (legacy settings.json)", path=legacy, config_exists=True + ) + ) + + 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 repair_config_file(path: str | os.PathLike) -> tuple[dict, list[str], str]: + """ + Attempt to repair a config file that failed strict parsing, WITHOUT writing + anything to disk. The caller decides what to do with the result (BCC shows + the user the issues and asks before saving). + + Returns (cfg, notes, repaired_text): + cfg - the parsed dict from the repaired JSON + notes - human-readable list of fixes that were applied + repaired_text - pretty-printed JSON the config would become + + Raises ValueError if the file can't be salvaged automatically. + """ + text = Path(path).read_text(encoding="utf-8") + candidate, notes = repair_json_text(text) + try: + obj = json.loads(candidate) + except json.JSONDecodeError as e: + raise ValueError( + f"Couldn't repair the file automatically (line {e.lineno}, column {e.colno}: {e.msg})." + ) from e + if not isinstance(obj, dict): + raise ValueError("Even after repair, the top level isn't a JSON object.") + pretty = json.dumps(obj, indent=2, ensure_ascii=False) + "\n" + return obj, notes, pretty + + +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]: + with contextlib.suppress(OSError): + old.unlink() + 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 + + +# --------------------------------------------------------------------------- # +# Backup utility functions +# --------------------------------------------------------------------------- # +_BACKUP_TS_RE = re.compile(r"(\d{8}-\d{6})") + + +def list_backups(config_path: Path | str) -> list[Path]: + """Return backup files for config_path, newest-first. Returns [] if none exist.""" + p = Path(config_path) + bdir = p.parent / BACKUP_DIRNAME + if not bdir.is_dir(): + return [] + return sorted(bdir.glob(f"{p.stem}.*.json"), reverse=True) + + +def backup_label(backup_path: Path | str) -> str: + """Human-readable label derived from the backup filename timestamp.""" + m = _BACKUP_TS_RE.search(Path(backup_path).name) + if not m: + return Path(backup_path).name + ts = m.group(1) # e.g. "20260702-004124" + return f"{ts[:4]}-{ts[4:6]}-{ts[6:8]} {ts[9:11]}:{ts[11:13]}:{ts[13:]}" + + +def _redact_server_data(data: dict) -> dict: + """Return a copy of a server definition with secrets masked for display.""" + out = dict(data) + if "args" in out: + out["args"] = redact_args(list(out["args"] or [])) + if "env" in out: + out["env"] = {k: (MASK if is_secret_key(k) else v) for k, v in (out["env"] or {}).items()} + return out + + +def _redact_servers_block(block: dict | None) -> dict: + """Return a sanitized copy of a mcpServers / _disabledMcpServers block.""" + if not block: + return {} + return {name: _redact_server_data(data) for name, data in block.items()} + + +def _server_sections(cfg: dict) -> dict: + """Return the masked server sections of a config dict, safe for diff display.""" + out: dict = {"mcpServers": _redact_servers_block(cfg.get("mcpServers"))} + if DISABLED_KEY in cfg: + out[DISABLED_KEY] = _redact_servers_block(cfg.get(DISABLED_KEY)) + return out + + +def backup_diff( + config_path: Path | str, + backup_path: Path | str, + current_cfg: dict | None = None, +) -> str: + """ + Unified diff showing what the server sections would look like after restoring + the backup. Only mcpServers and _disabledMcpServers are compared; non-server + keys are excluded entirely (they are preserved by restore, not changed). + Secret values in args and env are masked so the diff is safe to share. + + fromfile = current state, tofile = what will be written after restore. + Pass current_cfg to diff against an already-loaded in-memory config instead + of re-reading from disk. + """ + p = Path(config_path) + bp = Path(backup_path) + + if current_cfg is None: + try: + current_cfg = load_config(p) + except Exception: + current_cfg = {} + + backup_servers = extract_servers(load_config(bp)) + after_cfg = dict(current_cfg) + apply_servers(after_cfg, backup_servers) + + before_lines = ( + json.dumps(_server_sections(current_cfg), indent=2, ensure_ascii=False) + "\n" + ).splitlines(keepends=True) + after_lines = ( + json.dumps(_server_sections(after_cfg), indent=2, ensure_ascii=False) + "\n" + ).splitlines(keepends=True) + + result = "".join( + difflib.unified_diff( + before_lines, + after_lines, + fromfile=f"current: {p.name}", + tofile=f"restore: {bp.name}", + ) + ) + return result or "(no differences — backup matches current servers)" + + +def restore_backup( + config_path: Path | str, + backup_path: Path | str, + current_cfg: dict | None = None, +) -> Path | None: + """ + Restore server entries from backup_path into config_path. + + Only mcpServers and _disabledMcpServers are replaced; all other keys in the + current config (conversation history, project state, etc.) are preserved + verbatim and in their original order. + + Goes through write_config() so a pre-restore backup of the current file is + created automatically. Returns that backup path (or None if no prior file). + """ + p = Path(config_path) + if current_cfg is None: + try: + current_cfg = load_config(p) + except Exception: + current_cfg = {} + + backup_servers = extract_servers(load_config(Path(backup_path))) + cfg = dict(current_cfg) + apply_servers(cfg, backup_servers) + return write_config(p, cfg) + + +def config_mtime(path: Path | str) -> float | None: + """Return the file's mtime, or None if the file does not exist.""" + try: + return Path(path).stat().st_mtime + except OSError: + return None + + +class ConfigStat(NamedTuple): + """A snapshot of a config file's mtime + size. + + Pairing size with mtime hardens stale-file detection beyond bare mtime + equality: a concurrent external write can land within the filesystem's + mtime resolution (e.g. same-second writes on ext4/HFS+) or have its mtime + restored by the writing process, in which case mtime alone would miss the + change. Comparing both fields catches those cases without the cost of a + full content hash. + """ + + mtime: float + size: int + + +def config_fingerprint(path: Path | str) -> ConfigStat | None: + """Return the file's (mtime, size) snapshot, or None if it does not exist.""" + try: + st = Path(path).stat() + except OSError: + return None + return ConfigStat(st.st_mtime, st.st_size) + + +def external_change_summary(original_cfg: dict, path: Path | str) -> tuple[list[str], str]: + """ + Compare original_cfg (what BCC loaded) with the current on-disk state. + + Returns: + changed_keys — sorted list of top-level keys whose values differ. + server_diff — masked unified diff of server sections (empty if unchanged + or the file cannot be read). + """ + try: + disk_cfg = load_config(Path(path)) + except Exception: + return [], "" + + all_keys = set(original_cfg) | set(disk_cfg) + changed_keys = sorted(k for k in all_keys if original_cfg.get(k) != disk_cfg.get(k)) + + server_diff = "" + if any(k in {"mcpServers", DISABLED_KEY} for k in changed_keys): + before_lines = ( + json.dumps(_server_sections(original_cfg), indent=2, ensure_ascii=False) + "\n" + ).splitlines(keepends=True) + after_lines = ( + json.dumps(_server_sections(disk_cfg), indent=2, ensure_ascii=False) + "\n" + ).splitlines(keepends=True) + server_diff = "".join( + difflib.unified_diff(before_lines, after_lines, fromfile="loaded", tofile="on disk now") + ) + + return changed_keys, server_diff + + +# --------------------------------------------------------------------------- # +# 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" + + +# --------------------------------------------------------------------------- # +# Lenient JSON repair +# +# Snippets pasted from MCP docs, blog posts, and chat windows are frequently +# not valid JSON: markdown fences, surrounding prose, // comments, trailing +# commas, smart quotes, single quotes, unquoted keys, missing braces. The +# whole point of BCC is that nobody should have to hand-fix JSON, so the +# paste pipeline repairs what it can and reports what it changed. +# --------------------------------------------------------------------------- # + +# Word-processor / web artifacts mapped back to ASCII. +_QUOTE_MAP = { + "“": '"', + "”": '"', + "„": '"', + "«": '"', + "»": '"', + "‘": "'", + "’": "'", + "‚": "'", +} +_JUNK_CHARS = "​‌‍" # zero-width chars / BOM + + +def _strip_fences(text: str, notes: list[str]) -> str: + """Pull the contents out of a ```json ... ``` fence (or drop stray fence lines).""" + m = re.search(r"```(?:json[c5]?|javascript|js)?\s*\n(.*?)```", text, re.DOTALL | re.IGNORECASE) + if m: + notes.append("extracted code from markdown fence") + return m.group(1) + if "```" in text: + notes.append("removed markdown fence markers") + return text.replace("```json", "").replace("```", "") + return text + + +def _normalize_unicode(text: str, notes: list[str]) -> str: + out = text + for junk in _JUNK_CHARS: + out = out.replace(junk, "") + out = out.replace(chr(0xA0), " ") # non-breaking space (defensive: avoid a literal char here) + for smart, ascii_q in _QUOTE_MAP.items(): + out = out.replace(smart, ascii_q) + if out != text: + notes.append("normalized smart quotes / invisible characters") + return out + + +def _tokenize(text: str, notes: list[str]) -> list[tuple[bool, str]]: + """ + One careful pass over the text producing (is_string, segment) pairs. + While walking it also: converts single-quoted strings to double-quoted, + and drops //, /* */ and # comments (never inside strings). + """ + segs: list[tuple[bool, str]] = [] + buf: list[str] = [] + i, n = 0, len(text) + + def flush(): + if buf: + segs.append((False, "".join(buf))) + buf.clear() + + while i < n: + ch = text[i] + if ch == '"': # proper double-quoted string: copy verbatim + flush() + j = i + 1 + s = ['"'] + while j < n: + c = text[j] + s.append(c) + if c == "\\" and j + 1 < n: + s.append(text[j + 1]) + j += 2 + continue + if c == '"': + j += 1 + break + j += 1 + segs.append((True, "".join(s))) + i = j + elif ch == "'": # single-quoted string: convert to double-quoted + flush() + j = i + 1 + inner: list[str] = [] + while j < n: + c = text[j] + if c == "\\" and j + 1 < n: + nxt = text[j + 1] + if nxt == "'": + inner.append("'") # \' has no meaning in JSON + else: + inner.append(c) + inner.append(nxt) + j += 2 + continue + if c == "'": + j += 1 + break + if c == '"': + inner.append('\\"') + j += 1 + continue + inner.append(c) + j += 1 + segs.append((True, '"' + "".join(inner) + '"')) + notes.append("converted single-quoted strings") + i = j + elif ch == "/" and i + 1 < n and text[i + 1] == "/": + notes.append("removed // comments") + while i < n and text[i] != "\n": + i += 1 + elif ch == "/" and i + 1 < n and text[i + 1] == "*": + notes.append("removed /* */ comments") + i += 2 + while i + 1 < n and not (text[i] == "*" and text[i + 1] == "/"): + i += 1 + i = min(i + 2, n) + elif ch == "#": + # Only treat as a comment at line start (after whitespace) — never + # mid-value, where # could be part of an unquoted token. + line_start = text.rfind("\n", 0, i) + 1 + if text[line_start:i].strip() == "": + notes.append("removed # comments") + while i < n and text[i] != "\n": + i += 1 + else: + buf.append(ch) + i += 1 + else: + buf.append(ch) + i += 1 + flush() + return segs + + +def _repair_segments(segs: list[tuple[bool, str]], notes: list[str]) -> str: + """Structural fixes that only apply outside strings.""" + fixed: list[str] = [] + for idx, (is_str, seg) in enumerate(segs): + if is_str: + fixed.append(seg) + continue + s = seg + # Python / JS literals -> JSON + s2 = re.sub(r"\bTrue\b", "true", s) + s2 = re.sub(r"\bFalse\b", "false", s2) + s2 = re.sub(r"\bNone\b|\bundefined\b", "null", s2) + if s2 != s: + notes.append("converted Python/JS literals (True/False/None)") + s = s2 + # Unquoted keys: { key: or , key: -> "key": + s2 = re.sub(r"([{,]\s*)([A-Za-z_$][\w$.-]*)(\s*:)", r'\1"\2"\3', s) + if s2 != s: + notes.append("quoted bare object keys") + s = s2 + # Missing comma before the next string on a new line, e.g. + # "args": ["x"]\n"env": {...} or {...}\n"name2": {...} + # A string directly after a value-ish ending across a newline can never + # be valid JSON without a comma, so inserting one is always safe. + if idx + 1 < len(segs) and segs[idx + 1][0]: + body = s.rstrip() + tail_ws = s[len(body) :] + prev = body[-1:] if body else (fixed[-1].rstrip()[-1:] if fixed else "") + value_ending = prev in ('"', "}", "]") or prev.isdigit() + if "\n" in tail_ws and value_ending: + notes.append("inserted missing commas") + s = body + "," + tail_ws + fixed.append(s) + + out = "".join(fixed) + # Trailing commas (safe now: strings are intact, commas here are structural) + segs2 = _tokenize(out, []) + parts: list[str] = [] + changed = False + for is_str, seg in segs2: + if is_str: + parts.append(seg) + else: + new = re.sub(r",(\s*[}\]])", r"\1", seg) + changed = changed or new != seg + parts.append(new) + if changed: + notes.append("removed trailing commas") + return "".join(parts) + + +def _extract_and_balance(text: str, notes: list[str]) -> str: + """Slice out the JSON object (dropping surrounding prose) and close any + unclosed braces/brackets.""" + start = text.find("{") + if start == -1: + return text + if text[:start].strip(): + notes.append("ignored text before the JSON block") + + stack: list[str] = [] + i, n = start, len(text) + end = -1 + while i < n: + ch = text[i] + if ch == '"': # skip strings + i += 1 + while i < n: + if text[i] == "\\": + i += 2 + continue + if text[i] == '"': + break + i += 1 + elif ch in "{[": + stack.append("}" if ch == "{" else "]") + elif ch in "}]": + if stack and stack[-1] == ch: + stack.pop() + if not stack: + end = i + 1 + break + i += 1 + + if end != -1: + if text[end:].strip(): + notes.append("ignored text after the JSON block") + return text[start:end] + # Ran out of input with open scopes: close them. + if stack: + notes.append("closed unclosed braces/brackets") + return text[start:] + "".join(reversed(stack)) + return text[start:] + + +def repair_json_text(text: str) -> tuple[str, list[str]]: + """ + Best-effort repair of an almost-JSON snippet. Returns (candidate, notes) + where notes is a human-readable list of the fixes applied (deduplicated, + in order). Does NOT guarantee the result parses — callers still try + json.loads and surface its error if repair wasn't enough. + """ + notes: list[str] = [] + t = _strip_fences(text, notes) + t = _normalize_unicode(t, notes) + t = t.strip() + + # Brace-less inner fragment: "name": { ... } (with no outer braces). + # A quoted key at the start is a strong signal; a bare word only counts + # when followed by '{' so prose like "Note: ..." isn't swallowed. + if re.match(r'^\s*"[^"\n]+"\s*:\s*[{["]', t) or re.match(r"^\s*[A-Za-z_$][\w$.-]*\s*:\s*\{", t): + notes.append("wrapped fragment in braces") + t = "{" + t + "}" + + t = _extract_and_balance(t, notes) + segs = _tokenize(t, notes) + t = _repair_segments(segs, notes) + + # Re-balance in case comment/quote fixes exposed structure. + t = _extract_and_balance(t, []) + + seen: set[str] = set() + unique = [x for x in notes if not (x in seen or seen.add(x))] + return t, unique + + +def parse_pasted_json_verbose(text: str) -> tuple[dict[str, dict], list[str]]: + """ + Like parse_pasted_json, but forgiving. Tries strict JSON first; if that + fails, runs repair_json_text() and retries. Returns (servers, repair_notes). + repair_notes is empty when the input was already valid. + """ + raw = (text or "").strip() + if not raw: + raise ValueError("Nothing to parse.") + try: + return _shape_servers(json.loads(raw)), [] + except json.JSONDecodeError as strict_err: + candidate, notes = repair_json_text(raw) + try: + obj = json.loads(candidate) + except json.JSONDecodeError: + # Repair wasn't enough — report the original, more meaningful error. + raise ValueError( + f"Couldn't parse that as JSON even after auto-repair " + f"(line {strict_err.lineno}, column {strict_err.colno}: {strict_err.msg})." + ) from strict_err + return _shape_servers(obj), notes + + +def _shape_servers(obj) -> dict[str, dict]: + """Shared shape-detection for parsed paste content.""" + 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 (it needs a 'command' or a 'url')." + ) + clean[str(name)] = data + return clean + + +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) + + Input doesn't have to be valid JSON — markdown fences, comments, trailing + commas, smart/single quotes, unquoted keys, surrounding prose, and missing + braces are repaired automatically (see repair_json_text). + + Raises ValueError with a human message on anything unrecognizable. + """ + servers, _notes = parse_pasted_json_verbose(text) + return servers + + +# --------------------------------------------------------------------------- # +# Secrets: detection + redaction +# --------------------------------------------------------------------------- # +_SECRET_KEY_RE = re.compile( + r"(?i)(token|secret|passw|api[-_]?key|apikey|auth|credential|bearer|private[-_]?key|access[-_]?key)" +) + +# Well-known token prefixes that identify a bare value as a secret even +# without a telling key/flag name next to it. +_TOKEN_PREFIXES = ( + "ghp_", + "gho_", + "ghu_", + "ghs_", + "github_pat_", # GitHub + "sk-", + "sk_live_", + "sk_test_", + "rk_live_", # OpenAI / Stripe + "xoxb-", + "xoxp-", + "xoxc-", + "xoxs-", + "xapp-", # Slack + "glpat-", + "gldt-", # GitLab + "AKIA", + "ASIA", # AWS access key IDs + "ya29.", + "AIza", # Google + "pypi-", + "npm_", + "dop_v1_", + "figd_", +) + +MASK = "••••••••" + +# Matches userinfo credentials embedded in a URL: scheme://user:pass@host +# Fires on postgres://user:pass@host but NOT on https://host/path or ssh://user@host. +_EMBEDDED_CRED_RE = re.compile(r"://[^:@/\s]+:[^:@/\s]+@") + + +def is_secret_key(name: str) -> bool: + """Does this env-var / header / flag name look like it holds a secret?""" + return bool(_SECRET_KEY_RE.search(name or "")) + + +def _is_secret_value(value: str) -> bool: + return isinstance(value, str) and value.startswith(_TOKEN_PREFIXES) + + +def redact_args(args: list[str]) -> list[str]: + """ + Mask secret values in an args list for display/diagnostics: + --token abc123 -> --token •••••••• (value after a secret flag) + --api-key=abc123 -> --api-key=•••••••• (inline flag=value) + ghp_abc123 -> •••••••• (well-known token prefix) + Everything else passes through untouched. + """ + out: list[str] = [] + mask_next = False + for a in args: + s = str(a) + if mask_next: + out.append(MASK) + mask_next = False + continue + if s.startswith("-") and "=" in s and is_secret_key(s.split("=", 1)[0]): + out.append(s.split("=", 1)[0] + "=" + MASK) + continue + if s.startswith("-") and is_secret_key(s): + out.append(s) + mask_next = True + continue + if _is_secret_value(s): + out.append(MASK) + continue + out.append(s) + return out + + +def args_secret_warning(data: dict) -> str | None: + """ + Return a warning string when any arg looks like a raw secret that would + be better placed in `env`. Returns None when no concern is found. + + Skips --flag=value inline pairs (already partially self-documenting). + Fires on: + - positional values that start with a well-known token prefix (ghp_, sk-, …) + - values that follow a secret-named flag (--token abc, --api-key abc) + - URLs with embedded user:pass credentials (postgres://user:pass@host) + """ + args = [str(a) for a in (data.get("args") or [])] + mask_next = False + for a in args: + if mask_next: + mask_next = False + if not a.startswith("-"): + return ( + "An arg value following a secret-named flag looks like a credential. " + "Where the server supports it, prefer Environment variables — " + "args are visible in process listings." + ) + continue + # --flag=value inline: skip (the flag name already labels it) + if a.startswith("-") and "=" in a: + continue + # --secretflag (no inline value): flag the next positional arg + if a.startswith("-") and is_secret_key(a): + mask_next = True + continue + if a.startswith("-"): + continue + # Positional value: check for token prefix or embedded URL credentials + if _is_secret_value(a) or _EMBEDDED_CRED_RE.search(a): + return ( + "An arg value looks like a credential. " + "Where the server supports it, prefer Environment variables — " + "args are visible in process listings." + ) + return None + + +def split_suspicious_args(args: list[str]) -> tuple[list[str], list[str]]: + """ + Detect the classic argument-entry mistake: several argv tokens typed on one + line ("--directory /path/to/server"). An entry is only flagged when it + contains whitespace AND at least one whitespace-separated token starts with + "-" — so legitimate single arguments with spaces ("My Project Notes", + "/Users/me/My Documents") are never touched. + + Returns (fixed_args, notes). notes is empty when nothing was suspicious; + otherwise it describes each split so a UI can show the proposed fix. + Quotes are respected when splitting ('--name "My Server"' becomes + ['--name', 'My Server']). + """ + import shlex + + out: list[str] = [] + notes: list[str] = [] + for a in args: + if isinstance(a, str) and _looks_like_multiple_args(a): + try: + parts = shlex.split(a) + except ValueError: # unbalanced quotes — fall back to plain split + parts = a.split() + if len(parts) > 1: + out.extend(parts) + notes.append(f"“{a}” looks like {len(parts)} arguments — split onto separate lines") + continue + out.append(a) + return out, notes + + +def _looks_like_multiple_args(a: str) -> bool: + toks = a.split() + return len(toks) > 1 and any(t.startswith("-") for t in toks) + + +# --------------------------------------------------------------------------- # +# 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 system_path() -> str: + """ + The PATH that a GUI app (e.g., Claude Desktop launched from Finder/dock) + reliably inherits, independent of how BCC itself was started. + + This is used to determine the 'ok' vs 'warn' distinction: + ok = found in system_path() → works however Claude is launched + warn = found only in augmented_path() → works from a terminal but may + not work when Claude is opened from the desktop + + macOS : /etc/paths + /etc/paths.d/* (what launchd provides) plus + well-known package-manager prefixes (/opt/homebrew, /opt/local). + Windows: system + user PATH from the registry. + Linux : /etc/environment + standard FHS dirs + /snap/bin. + """ + dirs: list[str] = [] + + if sys.platform == "darwin": + for p in ["/etc/paths", *sorted(glob.glob("/etc/paths.d/*"))]: + with contextlib.suppress(OSError): + dirs.extend(ln.strip() for ln in Path(p).read_text().splitlines() if ln.strip()) + # Package-manager install prefixes: present on disk once installed, + # accessible to all processes regardless of shell configuration. + for d in ( + "/opt/homebrew/bin", + "/opt/homebrew/sbin", # Homebrew (Apple Silicon) + "/usr/local/bin", + "/usr/local/sbin", # Homebrew (Intel) / manual installs + "/opt/local/bin", + "/opt/local/sbin", # MacPorts + ): + dirs.append(d) + + elif os.name == "nt": + try: + import winreg + + for hive, sub in [ + ( + winreg.HKEY_LOCAL_MACHINE, + r"SYSTEM\CurrentControlSet\Control\Session Manager\Environment", + ), + (winreg.HKEY_CURRENT_USER, r"Environment"), + ]: + try: + with winreg.OpenKey(hive, sub) as k: + val, _ = winreg.QueryValueEx(k, "PATH") + dirs.extend(os.path.expandvars(val).split(os.pathsep)) + except OSError: + pass + except ImportError: + pass + dirs.extend( + [ + "C:\\Windows\\System32", + "C:\\Windows", + "C:\\Windows\\System32\\Wbem", + os.path.expandvars(r"%ProgramFiles%\nodejs"), + ] + ) + + else: # Linux / other + try: + for line in Path("/etc/environment").read_text().splitlines(): + if line.upper().startswith("PATH="): + dirs.extend(line[5:].strip("\"'").split(os.pathsep)) + except OSError: + pass + for d in ("/usr/local/bin", "/usr/bin", "/bin", "/usr/sbin", "/sbin", "/snap/bin"): + dirs.append(d) + + seen, out = set(), [] + for d in dirs: + d = d.strip() + if d and d not in seen: + seen.add(d) + out.append(d) + return os.pathsep.join(out) + + +@functools.lru_cache(maxsize=1) +def augmented_path() -> str: + """ + The widest PATH BCC searches when looking for a command — system_path() + plus user-specific runtime locations (nvm, cargo, volta, bun, etc.) that + require shell configuration to be active. + + Memoized for the session. Call refresh_path_cache() to re-scan, e.g. + after the user installs a new runtime. + """ + base = system_path().split(os.pathsep) + # Also include the current process PATH (catches unusual CI / container setups) + env_parts = os.environ.get("PATH", "").split(os.pathsep) + home = Path.home() + user_dirs = [ + str(home / ".local" / "bin"), + str(home / ".cargo" / "bin"), + str(home / ".deno" / "bin"), + str(home / ".bun" / "bin"), + str(home / ".volta" / "bin"), + ] + user_dirs += glob.glob(str(home / ".nvm" / "versions" / "node" / "*" / "bin")) + user_dirs += glob.glob( + str(home / ".local" / "share" / "fnm" / "node-versions" / "*" / "installation" / "bin") + ) + if os.name == "nt": + appdata = os.environ.get("APPDATA", "") + if appdata: + user_dirs.append(str(Path(appdata) / "npm")) + + seen, ordered = set(), [] + for p in base + env_parts + user_dirs: + 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 memoized PATHs so the next check re-scans (e.g. after an install).""" + system_path.cache_clear() + 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' means: found in system_path() — the PATH that Claude + # Desktop (or any GUI app) reliably inherits regardless of shell config. + # This check is consistent whether BCC runs from a terminal or as a .app. + in_base[c] = bool(shutil.which(c, path=system_path())) + + 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 that location is not in the standard " + f"system PATH. Claude Desktop launched from Finder or the dock may not see it. " + f"Clicking 'Use full path ↳' rewrites the command to its absolute path, which " + f"works regardless of how Claude is launched." + ) + 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"): + # Redacted: this report is designed to be pasted into bug reports. + L.append("Args: " + " ".join(redact_args(list(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}") + + sys_dirs = set(system_path().split(os.pathsep)) + aug_dirs = res["searched_path"].split(os.pathsep) if res["searched_path"] else [] + extra = [d for d in aug_dirs if d not in sys_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( + "(+ = user-specific dirs not in the standard system PATH. " + "Claude Desktop launched from Finder may NOT have these.)" + ) + return "\n".join(L) + + +_STDERR_CAP = 4096 # bytes + + +def spawn_test(data: dict, timeout: float = 3.0) -> dict: + """ + Attempt to start a stdio server and observe it for `timeout` seconds. + + Returns a dict: + outcome: "ok" — still running after timeout; server started correctly + "exited" — exited with code 0 before timeout; unusual for a server + "crashed" — exited with a non-zero code before timeout + "not_found" — command could not be resolved to an executable + "not_applicable" — remote server or no command; nothing to spawn + returncode: int | None + stderr: str (first ~4 KB) + detail: str + + Run this off the UI thread — it blocks for up to `timeout` seconds. + """ + if "url" in data and "command" not in data: + return { + "outcome": "not_applicable", + "returncode": None, + "stderr": "", + "detail": "remote server", + } + + cmd = (data.get("command") or "").strip() + if not cmd: + return { + "outcome": "not_applicable", + "returncode": None, + "stderr": "", + "detail": "no command set", + } + + # Resolve to absolute path so subprocess doesn't fight with PATH in env. + resolved_cmd = shutil.which(cmd, path=augmented_path()) + if resolved_cmd is None: + return { + "outcome": "not_found", + "returncode": None, + "stderr": "", + "detail": f"command not found: {cmd}", + } + + args_list = [resolved_cmd] + [str(a) for a in (data.get("args") or [])] + + merged_env = {**os.environ, "PATH": augmented_path()} + merged_env.update(data.get("env") or {}) + + stderr_chunks: list[bytes] = [] + + def _drain(pipe) -> None: + # Read until EOF so the pipe buffer never fills and blocks the subprocess. + # Only keep the first _STDERR_CAP bytes; the rest is discarded. + captured = 0 + try: + while True: + chunk = pipe.read(1024) + if not chunk: + break + if captured < _STDERR_CAP: + keep = min(len(chunk), _STDERR_CAP - captured) + stderr_chunks.append(chunk[:keep]) + captured += keep + except Exception: + pass + + popen_kwargs: dict = dict( + stdin=subprocess.PIPE, # keep open so servers block on read rather than seeing EOF + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + env=merged_env, + ) + if os.name != "nt": + popen_kwargs["start_new_session"] = True # own process group → clean kill + + try: + proc = subprocess.Popen(args_list, **popen_kwargs) + except (FileNotFoundError, OSError) as e: + return { + "outcome": "not_found", + "returncode": None, + "stderr": "", + "detail": str(e), + } + + drain_thread = threading.Thread(target=_drain, args=(proc.stderr,), daemon=True) + drain_thread.start() + + try: + proc.wait(timeout=timeout) + except subprocess.TimeoutExpired: + try: + if os.name != "nt": + os.killpg(os.getpgid(proc.pid), _signal.SIGKILL) + else: + proc.kill() # best-effort on Windows + except OSError: + pass + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=1.0) + drain_thread.join(0.5) + stderr = b"".join(stderr_chunks).decode(errors="replace") + return { + "outcome": "ok", + "returncode": None, + "stderr": stderr, + "detail": f"still running after {timeout:.0f}s — server started successfully", + } + + drain_thread.join(0.5) + stderr = b"".join(stderr_chunks).decode(errors="replace") + rc = proc.returncode + if rc == 0: + return { + "outcome": "exited", + "returncode": rc, + "stderr": stderr, + "detail": "process exited cleanly (code 0) — unusual; a healthy server should keep running", + } + return { + "outcome": "crashed", + "returncode": rc, + "stderr": stderr, + "detail": f"process exited with code {rc}", + } + + +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 TimeoutError: + return False, "timed out" + except Exception as e: + 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