the_og
9760b1537e
The previous commit accidentally normalized the non-breaking space (U+00A0) in _normalize_unicode's replace() call to a regular space during a copy/paste, turning that replace() into a no-op. Use an explicit escape instead of the literal character so it can't be silently corrupted again.
1487 lines
53 KiB
Python
1487 lines
53 KiB
Python
"""
|
||
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/<version>/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
|