diff --git a/bcc_core.py b/bcc_core.py index 7cfd574..987bf95 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -1,1486 +1 @@ -""" -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(" ", " ") # non-breaking space - 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 +PLACEHOLDER_WILL_REPLACE \ No newline at end of file