Cowork Supervisor
c493aa0c84
- Linux (and any non-desktop platform): refuse instead of 'pkill claude', which substring-matched running Claude Code CLI sessions and relaunched the CLI, not a desktop app. New restart_supported() gates the button. - macOS: wait (<=5s) for the old instance to exit before 'open -a Claude' so the relaunch can't re-activate the dying process. Runs off the UI thread via a RestartWorker. - Windows: verify the Start-menu shortcut exists BEFORE taskkill, so an MSIX/Store install is never killed without a relaunch path. Closes #33 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1935 lines
71 KiB
Python
1935 lines
71 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"}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Version / update checking
|
||
#
|
||
# __version__ is the single source of truth for the app version (must match
|
||
# pyproject.toml's [project] version). The About dialog and the update
|
||
# checker both read this constant instead of hard-coding a version string.
|
||
#
|
||
# The update checker is notify-only: it reads release metadata from the
|
||
# repo's Gitea releases API and NEVER downloads or replaces the running
|
||
# binary. All network I/O here is fail-quiet (returns None on any problem)
|
||
# so it's safe to run unattended, off the UI thread, at startup.
|
||
# --------------------------------------------------------------------------- #
|
||
__version__ = "1.2.0"
|
||
|
||
REPO_URL = "https://git.avezzano.io/the_og/better-claude-config"
|
||
ISSUES_URL = f"{REPO_URL}/issues"
|
||
RELEASES_URL = f"{REPO_URL}/releases"
|
||
LICENSE_URL = f"{REPO_URL}/raw/branch/main/LICENSE"
|
||
|
||
# Public repo -> anonymously reachable, no auth/token needed or embedded.
|
||
_RELEASES_API_URL = (
|
||
"https://git.avezzano.io/api/v1/repos/the_og/better-claude-config/releases/latest"
|
||
)
|
||
|
||
|
||
def parse_version(v: str) -> tuple[int, ...]:
|
||
"""
|
||
Parse a version string into a tuple of ints for numeric comparison.
|
||
|
||
Strips a leading 'v' ("v1.2.3" -> "1.2.3") and any pre-release/build
|
||
metadata after a '-' or '+' ("1.2.3-beta.1" -> "1.2.3"). Stops at the
|
||
first non-numeric dotted component. Empty or entirely non-numeric input
|
||
returns an empty tuple rather than raising, so a malformed tag from a
|
||
flaky API response degrades gracefully instead of crashing the caller.
|
||
"""
|
||
s = (v or "").strip()
|
||
if s[:1].lower() == "v":
|
||
s = s[1:]
|
||
s = re.split(r"[-+]", s, maxsplit=1)[0]
|
||
parts: list[int] = []
|
||
for chunk in s.split("."):
|
||
m = re.match(r"\d+", chunk)
|
||
if not m:
|
||
break
|
||
parts.append(int(m.group()))
|
||
return tuple(parts)
|
||
|
||
|
||
def is_newer_version(current: str, candidate: str) -> bool:
|
||
"""
|
||
True if `candidate` is a strictly newer version than `current`.
|
||
|
||
Comparison is purely numeric (major.minor.patch, ...) — NEVER a lexical
|
||
string compare, so "v2.0.0" vs "v10.0.0" resolves correctly instead of
|
||
sorting "2" after "10". Tuples of differing length are zero-padded before
|
||
comparing, so "1.2" and "1.2.0" are correctly treated as equal.
|
||
|
||
An unparseable `candidate` always yields False (nothing to report). An
|
||
unparseable `current` is treated as "0" for comparison purposes — a
|
||
malformed local version shouldn't silently suppress a real update.
|
||
"""
|
||
cur = parse_version(current)
|
||
new = parse_version(candidate)
|
||
if not new:
|
||
return False
|
||
width = max(len(cur), len(new), 1)
|
||
cur = cur + (0,) * (width - len(cur))
|
||
new = new + (0,) * (width - len(new))
|
||
return new > cur
|
||
|
||
|
||
def fetch_latest_release(timeout: float = 4.0) -> dict | None:
|
||
"""
|
||
Query the repo's (public, anonymous) Gitea releases API for the latest
|
||
release. Returns {"version": "<tag>", "url": "<releases page>"} on
|
||
success, or None on ANY failure: network error, timeout, bad status,
|
||
malformed JSON, or a response missing tag_name.
|
||
|
||
Fail-quiet by design — this is meant to be called off the UI thread
|
||
(see UpdateCheckWorker in bcc.py) for both the About dialog's "Check for
|
||
updates" button and an optional silent startup check. Never downloads or
|
||
touches any binary; this only ever reads release metadata.
|
||
"""
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
req = urllib.request.Request(
|
||
_RELEASES_API_URL,
|
||
headers={
|
||
"Accept": "application/json",
|
||
"User-Agent": f"BetterClaudeConfig/{__version__}",
|
||
},
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
payload = json.loads(r.read().decode("utf-8"))
|
||
except (urllib.error.URLError, TimeoutError, ValueError, OSError):
|
||
return None
|
||
if not isinstance(payload, dict):
|
||
return None
|
||
tag = payload.get("tag_name")
|
||
if not tag or not isinstance(tag, str):
|
||
return None
|
||
return {"version": tag, "url": payload.get("html_url") or RELEASES_URL}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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 msix_config_paths(localappdata: str | os.PathLike | None = None) -> list[Path]:
|
||
"""
|
||
Find MSIX/Store-packaged Claude Desktop configs.
|
||
|
||
When Claude Desktop is installed from the Microsoft Store (MSIX), Windows
|
||
virtualizes its filesystem writes to a per-package folder under
|
||
`%LOCALAPPDATA%\\Packages\\<PackageFamilyName>\\LocalCache\\Roaming\\Claude\\`
|
||
instead of the normal `%APPDATA%\\Claude\\`. A user (or BCC) editing the
|
||
plain %APPDATA% path can end up changing a file the running app never
|
||
reads -- see anthropics/claude-code issues #26073, #29100, #38830.
|
||
|
||
Globs `<localappdata>/Packages/*Claude*/LocalCache/Roaming/Claude/
|
||
claude_desktop_config.json` and returns every match that actually exists,
|
||
sorted for determinism. `localappdata` defaults to the %LOCALAPPDATA% env
|
||
var (falling back to the usual Windows path) but is accepted as a
|
||
parameter so this is unit-testable with tmp_path on any platform.
|
||
|
||
This function itself is platform-independent (it just globs whatever
|
||
directory it's given); callers that care about the *current* machine
|
||
should gate on sys.platform -- see `detect_msix_claude`.
|
||
"""
|
||
base = (
|
||
Path(localappdata)
|
||
if localappdata is not None
|
||
else Path(os.environ.get("LOCALAPPDATA", str(Path.home() / "AppData" / "Local")))
|
||
)
|
||
packages = base / "Packages"
|
||
if not packages.is_dir():
|
||
return []
|
||
out: list[Path] = []
|
||
for pkg_dir in sorted(packages.glob("*Claude*")):
|
||
cfg = pkg_dir / "LocalCache" / "Roaming" / "Claude" / CONFIG_FILENAME
|
||
if cfg.is_file():
|
||
out.append(cfg)
|
||
return out
|
||
|
||
|
||
def detect_msix_claude(
|
||
appdata: str | os.PathLike | None = None,
|
||
localappdata: str | os.PathLike | None = None,
|
||
) -> Path | None:
|
||
"""
|
||
Best-effort detection of an MSIX-virtualized Claude Desktop install.
|
||
|
||
Returns the first virtualized `claude_desktop_config.json` found (see
|
||
`msix_config_paths`), or None when not running on Windows, no matching
|
||
package folder exists, or a package folder exists but has no config file
|
||
written yet. The sys.platform gate makes this a safe no-op to call
|
||
unconditionally from discovery/diagnostics code on macOS/Linux.
|
||
|
||
`appdata`/`localappdata` are threaded through (rather than read straight
|
||
from os.environ) purely so the whole detection path is unit-testable via
|
||
tmp_path + monkeypatch without mutating real env vars.
|
||
"""
|
||
if not sys.platform.startswith("win"):
|
||
return None
|
||
hits = msix_config_paths(localappdata)
|
||
return hits[0] if hits else None
|
||
|
||
|
||
def msix_warning_text(
|
||
appdata: str | os.PathLike | None = None,
|
||
localappdata: str | os.PathLike | None = None,
|
||
) -> str | None:
|
||
"""
|
||
A one-line, paste-safe warning for the diagnostics/status surface when
|
||
Claude Desktop looks like an MSIX/Store install whose real config lives
|
||
somewhere other than the plain %APPDATA%\\Claude\\ path. Returns None
|
||
when nothing was detected (including on non-Windows platforms) or when
|
||
the virtualized path and the plain path happen to coincide -- i.e. there
|
||
is nothing surprising to warn about. Contains only filesystem paths, no
|
||
env values or secrets.
|
||
"""
|
||
real = detect_msix_claude(appdata, localappdata)
|
||
if real is None:
|
||
return None
|
||
plain_base = (
|
||
Path(appdata)
|
||
if appdata is not None
|
||
else Path(os.environ.get("APPDATA", str(Path.home() / "AppData" / "Roaming")))
|
||
)
|
||
plain_cfg = plain_base / "Claude" / CONFIG_FILENAME
|
||
if plain_cfg == real:
|
||
return None
|
||
return (
|
||
"Claude Desktop looks like it's installed from the Microsoft Store (MSIX). "
|
||
f"Windows virtualizes its config, so edits to {plain_cfg} may be silently "
|
||
f"ignored by the running app. The real config is at: {real}"
|
||
)
|
||
|
||
|
||
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.).
|
||
Windows/MSIX: if Claude Desktop was installed from the Microsoft Store,
|
||
its real config lives in a virtualized per-package folder rather than the
|
||
plain %APPDATA%\\Claude\\ path above (see `detect_msix_claude`); when
|
||
that's detected, it's surfaced here as its own profile so the user can
|
||
edit the file the app actually reads.
|
||
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()))
|
||
|
||
msix_cfg = detect_msix_claude()
|
||
if msix_cfg is not None and str(msix_cfg) not in {str(p.path) for p in out}:
|
||
out.append(
|
||
Profile(label="Claude (Microsoft Store / MSIX)", path=msix_cfg, config_exists=True)
|
||
)
|
||
|
||
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())
|
||
|
||
|
||
def profile_targets_claude_desktop(profile: Profile) -> bool:
|
||
"""
|
||
True when `profile` points at a Claude Desktop config
|
||
(claude_desktop_config.json), as opposed to Claude Code (~/.claude.json
|
||
or the legacy ~/.claude/settings.json). Used to gate Desktop-only actions
|
||
like "Restart Claude Desktop" so they never show up for a Claude Code
|
||
profile -- restarting the CLI makes no sense.
|
||
"""
|
||
return Path(profile.path).name == CONFIG_FILENAME
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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 resolve_name_collision(name: str, existing: set[str]) -> str:
|
||
"""
|
||
Return a name guaranteed not to collide with `existing`.
|
||
|
||
If `name` isn't already taken it's returned unchanged. Otherwise a
|
||
`-2`, `-3`, ... suffix is appended until the result is unique — this is
|
||
the "keep both (renamed)" branch used by paste/import when the user
|
||
doesn't want to overwrite an existing server of the same name.
|
||
"""
|
||
if name not in existing:
|
||
return name
|
||
n = 2
|
||
candidate = f"{name}-{n}"
|
||
while candidate in existing:
|
||
n += 1
|
||
candidate = f"{name}-{n}"
|
||
return candidate
|
||
|
||
|
||
def apply_servers(cfg: dict, servers: list[ServerEntry]) -> dict:
|
||
"""
|
||
Write the server list back into `cfg` in place, preserving every other key
|
||
and the position of `mcpServers`. Returns the same dict for convenience.
|
||
"""
|
||
enabled = {s.name: s.data for s in servers if s.enabled}
|
||
disabled = {s.name: s.data for s in servers if not s.enabled}
|
||
|
||
cfg["mcpServers"] = enabled # replaces value if key existed; appends otherwise
|
||
if disabled:
|
||
cfg[DISABLED_KEY] = disabled
|
||
else:
|
||
cfg.pop(DISABLED_KEY, None)
|
||
return cfg
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Write (atomic, with rotating backups)
|
||
# --------------------------------------------------------------------------- #
|
||
def _make_backup(path: Path) -> Path:
|
||
bdir = path.parent / BACKUP_DIRNAME
|
||
bdir.mkdir(exist_ok=True)
|
||
stamp = time.strftime("%Y%m%d-%H%M%S")
|
||
dest = bdir / f"{path.stem}.{stamp}.json"
|
||
# Avoid clobbering a same-second backup
|
||
n = 1
|
||
while dest.exists():
|
||
dest = bdir / f"{path.stem}.{stamp}.{n}.json"
|
||
n += 1
|
||
shutil.copy2(path, dest)
|
||
# Prune oldest beyond MAX_BACKUPS
|
||
backups = sorted(bdir.glob(f"{path.stem}.*.json"))
|
||
for old in backups[:-MAX_BACKUPS]:
|
||
with contextlib.suppress(OSError):
|
||
old.unlink()
|
||
return dest
|
||
|
||
|
||
def write_config(path: str | os.PathLike, cfg: dict) -> Path | None:
|
||
"""
|
||
Atomically write `cfg` to `path` (2-space pretty JSON). Backs up any existing
|
||
file first. Returns the backup path (or None if there was nothing to back up).
|
||
"""
|
||
p = Path(path)
|
||
p.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
backup = _make_backup(p) if p.is_file() else None
|
||
|
||
payload = json.dumps(cfg, indent=2, ensure_ascii=False) + "\n"
|
||
fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=".tmp_mcp_", suffix=".json")
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||
f.write(payload)
|
||
os.replace(tmp, p) # atomic on the same filesystem
|
||
finally:
|
||
if os.path.exists(tmp):
|
||
os.remove(tmp)
|
||
return backup
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Backup utility functions
|
||
# --------------------------------------------------------------------------- #
|
||
_BACKUP_TS_RE = re.compile(r"(\d{8}-\d{6})")
|
||
|
||
|
||
def list_backups(config_path: Path | str) -> list[Path]:
|
||
"""Return backup files for config_path, newest-first. Returns [] if none exist."""
|
||
p = Path(config_path)
|
||
bdir = p.parent / BACKUP_DIRNAME
|
||
if not bdir.is_dir():
|
||
return []
|
||
return sorted(bdir.glob(f"{p.stem}.*.json"), reverse=True)
|
||
|
||
|
||
def backup_label(backup_path: Path | str) -> str:
|
||
"""Human-readable label derived from the backup filename timestamp."""
|
||
m = _BACKUP_TS_RE.search(Path(backup_path).name)
|
||
if not m:
|
||
return Path(backup_path).name
|
||
ts = m.group(1) # e.g. "20260702-004124"
|
||
return f"{ts[:4]}-{ts[4:6]}-{ts[6:8]} {ts[9:11]}:{ts[11:13]}:{ts[13:]}"
|
||
|
||
|
||
def _redact_server_data(data: dict) -> dict:
|
||
"""Return a copy of a server definition with secrets masked for display."""
|
||
out = dict(data)
|
||
if "args" in out:
|
||
out["args"] = redact_args(list(out["args"] or []))
|
||
if "env" in out:
|
||
out["env"] = {k: (MASK if is_secret_key(k) else v) for k, v in (out["env"] or {}).items()}
|
||
return out
|
||
|
||
|
||
def _redact_servers_block(block: dict | None) -> dict:
|
||
"""Return a sanitized copy of a mcpServers / _disabledMcpServers block."""
|
||
if not block:
|
||
return {}
|
||
return {name: _redact_server_data(data) for name, data in block.items()}
|
||
|
||
|
||
def _server_sections(cfg: dict) -> dict:
|
||
"""Return the masked server sections of a config dict, safe for diff display."""
|
||
out: dict = {"mcpServers": _redact_servers_block(cfg.get("mcpServers"))}
|
||
if DISABLED_KEY in cfg:
|
||
out[DISABLED_KEY] = _redact_servers_block(cfg.get(DISABLED_KEY))
|
||
return out
|
||
|
||
|
||
def backup_diff(
|
||
config_path: Path | str,
|
||
backup_path: Path | str,
|
||
current_cfg: dict | None = None,
|
||
) -> str:
|
||
"""
|
||
Unified diff showing what the server sections would look like after restoring
|
||
the backup. Only mcpServers and _disabledMcpServers are compared; non-server
|
||
keys are excluded entirely (they are preserved by restore, not changed).
|
||
Secret values in args and env are masked so the diff is safe to share.
|
||
|
||
fromfile = current state, tofile = what will be written after restore.
|
||
Pass current_cfg to diff against an already-loaded in-memory config instead
|
||
of re-reading from disk.
|
||
"""
|
||
p = Path(config_path)
|
||
bp = Path(backup_path)
|
||
|
||
if current_cfg is None:
|
||
try:
|
||
current_cfg = load_config(p)
|
||
except Exception:
|
||
current_cfg = {}
|
||
|
||
backup_servers = extract_servers(load_config(bp))
|
||
after_cfg = dict(current_cfg)
|
||
apply_servers(after_cfg, backup_servers)
|
||
|
||
before_lines = (
|
||
json.dumps(_server_sections(current_cfg), indent=2, ensure_ascii=False) + "\n"
|
||
).splitlines(keepends=True)
|
||
after_lines = (
|
||
json.dumps(_server_sections(after_cfg), indent=2, ensure_ascii=False) + "\n"
|
||
).splitlines(keepends=True)
|
||
|
||
result = "".join(
|
||
difflib.unified_diff(
|
||
before_lines,
|
||
after_lines,
|
||
fromfile=f"current: {p.name}",
|
||
tofile=f"restore: {bp.name}",
|
||
)
|
||
)
|
||
return result or "(no differences — backup matches current servers)"
|
||
|
||
|
||
def restore_backup(
|
||
config_path: Path | str,
|
||
backup_path: Path | str,
|
||
current_cfg: dict | None = None,
|
||
) -> Path | None:
|
||
"""
|
||
Restore server entries from backup_path into config_path.
|
||
|
||
Only mcpServers and _disabledMcpServers are replaced; all other keys in the
|
||
current config (conversation history, project state, etc.) are preserved
|
||
verbatim and in their original order.
|
||
|
||
Goes through write_config() so a pre-restore backup of the current file is
|
||
created automatically. Returns that backup path (or None if no prior file).
|
||
"""
|
||
p = Path(config_path)
|
||
if current_cfg is None:
|
||
try:
|
||
current_cfg = load_config(p)
|
||
except Exception:
|
||
current_cfg = {}
|
||
|
||
backup_servers = extract_servers(load_config(Path(backup_path)))
|
||
cfg = dict(current_cfg)
|
||
apply_servers(cfg, backup_servers)
|
||
return write_config(p, cfg)
|
||
|
||
|
||
def config_mtime(path: Path | str) -> float | None:
|
||
"""Return the file's mtime, or None if the file does not exist."""
|
||
try:
|
||
return Path(path).stat().st_mtime
|
||
except OSError:
|
||
return None
|
||
|
||
|
||
class ConfigStat(NamedTuple):
|
||
"""A snapshot of a config file's mtime + size.
|
||
|
||
Pairing size with mtime hardens stale-file detection beyond bare mtime
|
||
equality: a concurrent external write can land within the filesystem's
|
||
mtime resolution (e.g. same-second writes on ext4/HFS+) or have its mtime
|
||
restored by the writing process, in which case mtime alone would miss the
|
||
change. Comparing both fields catches those cases without the cost of a
|
||
full content hash.
|
||
"""
|
||
|
||
mtime: float
|
||
size: int
|
||
|
||
|
||
def config_fingerprint(path: Path | str) -> ConfigStat | None:
|
||
"""Return the file's (mtime, size) snapshot, or None if it does not exist."""
|
||
try:
|
||
st = Path(path).stat()
|
||
except OSError:
|
||
return None
|
||
return ConfigStat(st.st_mtime, st.st_size)
|
||
|
||
|
||
def external_change_summary(original_cfg: dict, path: Path | str) -> tuple[list[str], str]:
|
||
"""
|
||
Compare original_cfg (what BCC loaded) with the current on-disk state.
|
||
|
||
Returns:
|
||
changed_keys — sorted list of top-level keys whose values differ.
|
||
server_diff — masked unified diff of server sections (empty if unchanged
|
||
or the file cannot be read).
|
||
"""
|
||
try:
|
||
disk_cfg = load_config(Path(path))
|
||
except Exception:
|
||
return [], ""
|
||
|
||
all_keys = set(original_cfg) | set(disk_cfg)
|
||
changed_keys = sorted(k for k in all_keys if original_cfg.get(k) != disk_cfg.get(k))
|
||
|
||
server_diff = ""
|
||
if any(k in {"mcpServers", DISABLED_KEY} for k in changed_keys):
|
||
before_lines = (
|
||
json.dumps(_server_sections(original_cfg), indent=2, ensure_ascii=False) + "\n"
|
||
).splitlines(keepends=True)
|
||
after_lines = (
|
||
json.dumps(_server_sections(disk_cfg), indent=2, ensure_ascii=False) + "\n"
|
||
).splitlines(keepends=True)
|
||
server_diff = "".join(
|
||
difflib.unified_diff(before_lines, after_lines, fromfile="loaded", tofile="on disk now")
|
||
)
|
||
|
||
return changed_keys, server_diff
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Paste / import parsing
|
||
# --------------------------------------------------------------------------- #
|
||
def looks_like_server(data) -> bool:
|
||
return isinstance(data, dict) and ("command" in data or "url" in data)
|
||
|
||
|
||
def suggest_name(data: dict) -> str:
|
||
"""Derive a friendly default name from a server definition."""
|
||
if "url" in data and "command" not in data:
|
||
host = urlparse(data["url"]).hostname or "remote"
|
||
return host.split(".")[0] or "remote"
|
||
for a in data.get("args", []) or []:
|
||
if isinstance(a, str) and ("/" in a or a.startswith("@")):
|
||
base = a.rstrip("/").split("/")[-1]
|
||
base = base.replace("server-", "").replace("mcp-server-", "").replace("mcp-", "")
|
||
if base:
|
||
return base
|
||
cmd = data.get("command", "server")
|
||
return Path(str(cmd)).stem or "server"
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Lenient JSON repair
|
||
#
|
||
# Snippets pasted from MCP docs, blog posts, and chat windows are frequently
|
||
# not valid JSON: markdown fences, surrounding prose, // comments, trailing
|
||
# commas, smart quotes, single quotes, unquoted keys, missing braces. The
|
||
# whole point of BCC is that nobody should have to hand-fix JSON, so the
|
||
# paste pipeline repairs what it can and reports what it changed.
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
# Word-processor / web artifacts mapped back to ASCII.
|
||
_QUOTE_MAP = {
|
||
"“": '"',
|
||
"”": '"',
|
||
"„": '"',
|
||
"«": '"',
|
||
"»": '"',
|
||
"‘": "'",
|
||
"’": "'",
|
||
"‚": "'",
|
||
}
|
||
_JUNK_CHARS = "" # zero-width chars / BOM
|
||
|
||
|
||
def _strip_fences(text: str, notes: list[str]) -> str:
|
||
"""Pull the contents out of a ```json ... ``` fence (or drop stray fence lines)."""
|
||
m = re.search(r"```(?:json[c5]?|javascript|js)?\s*\n(.*?)```", text, re.DOTALL | re.IGNORECASE)
|
||
if m:
|
||
notes.append("extracted code from markdown fence")
|
||
return m.group(1)
|
||
if "```" in text:
|
||
notes.append("removed markdown fence markers")
|
||
return text.replace("```json", "").replace("```", "")
|
||
return text
|
||
|
||
|
||
def _normalize_unicode(text: str, notes: list[str]) -> str:
|
||
out = text
|
||
for junk in _JUNK_CHARS:
|
||
out = out.replace(junk, "")
|
||
out = out.replace(chr(0xA0), " ") # non-breaking space (defensive: avoid a literal char here)
|
||
for smart, ascii_q in _QUOTE_MAP.items():
|
||
out = out.replace(smart, ascii_q)
|
||
if out != text:
|
||
notes.append("normalized smart quotes / invisible characters")
|
||
return out
|
||
|
||
|
||
def _tokenize(text: str, notes: list[str]) -> list[tuple[bool, str]]:
|
||
"""
|
||
One careful pass over the text producing (is_string, segment) pairs.
|
||
While walking it also: converts single-quoted strings to double-quoted,
|
||
and drops //, /* */ and # comments (never inside strings).
|
||
"""
|
||
segs: list[tuple[bool, str]] = []
|
||
buf: list[str] = []
|
||
i, n = 0, len(text)
|
||
|
||
def flush():
|
||
if buf:
|
||
segs.append((False, "".join(buf)))
|
||
buf.clear()
|
||
|
||
while i < n:
|
||
ch = text[i]
|
||
if ch == '"': # proper double-quoted string: copy verbatim
|
||
flush()
|
||
j = i + 1
|
||
s = ['"']
|
||
while j < n:
|
||
c = text[j]
|
||
s.append(c)
|
||
if c == "\\" and j + 1 < n:
|
||
s.append(text[j + 1])
|
||
j += 2
|
||
continue
|
||
if c == '"':
|
||
j += 1
|
||
break
|
||
j += 1
|
||
segs.append((True, "".join(s)))
|
||
i = j
|
||
elif ch == "'": # single-quoted string: convert to double-quoted
|
||
flush()
|
||
j = i + 1
|
||
inner: list[str] = []
|
||
while j < n:
|
||
c = text[j]
|
||
if c == "\\" and j + 1 < n:
|
||
nxt = text[j + 1]
|
||
if nxt == "'":
|
||
inner.append("'") # \' has no meaning in JSON
|
||
else:
|
||
inner.append(c)
|
||
inner.append(nxt)
|
||
j += 2
|
||
continue
|
||
if c == "'":
|
||
j += 1
|
||
break
|
||
if c == '"':
|
||
inner.append('\\"')
|
||
j += 1
|
||
continue
|
||
inner.append(c)
|
||
j += 1
|
||
segs.append((True, '"' + "".join(inner) + '"'))
|
||
notes.append("converted single-quoted strings")
|
||
i = j
|
||
elif ch == "/" and i + 1 < n and text[i + 1] == "/":
|
||
notes.append("removed // comments")
|
||
while i < n and text[i] != "\n":
|
||
i += 1
|
||
elif ch == "/" and i + 1 < n and text[i + 1] == "*":
|
||
notes.append("removed /* */ comments")
|
||
i += 2
|
||
while i + 1 < n and not (text[i] == "*" and text[i + 1] == "/"):
|
||
i += 1
|
||
i = min(i + 2, n)
|
||
elif ch == "#":
|
||
# Only treat as a comment at line start (after whitespace) — never
|
||
# mid-value, where # could be part of an unquoted token.
|
||
line_start = text.rfind("\n", 0, i) + 1
|
||
if text[line_start:i].strip() == "":
|
||
notes.append("removed # comments")
|
||
while i < n and text[i] != "\n":
|
||
i += 1
|
||
else:
|
||
buf.append(ch)
|
||
i += 1
|
||
else:
|
||
buf.append(ch)
|
||
i += 1
|
||
flush()
|
||
return segs
|
||
|
||
|
||
def _repair_segments(segs: list[tuple[bool, str]], notes: list[str]) -> str:
|
||
"""Structural fixes that only apply outside strings."""
|
||
fixed: list[str] = []
|
||
for idx, (is_str, seg) in enumerate(segs):
|
||
if is_str:
|
||
fixed.append(seg)
|
||
continue
|
||
s = seg
|
||
# Python / JS literals -> JSON
|
||
s2 = re.sub(r"\bTrue\b", "true", s)
|
||
s2 = re.sub(r"\bFalse\b", "false", s2)
|
||
s2 = re.sub(r"\bNone\b|\bundefined\b", "null", s2)
|
||
if s2 != s:
|
||
notes.append("converted Python/JS literals (True/False/None)")
|
||
s = s2
|
||
# Unquoted keys: { key: or , key: -> "key":
|
||
s2 = re.sub(r"([{,]\s*)([A-Za-z_$][\w$.-]*)(\s*:)", r'\1"\2"\3', s)
|
||
if s2 != s:
|
||
notes.append("quoted bare object keys")
|
||
s = s2
|
||
# Missing comma before the next string on a new line, e.g.
|
||
# "args": ["x"]\n"env": {...} or {...}\n"name2": {...}
|
||
# A string directly after a value-ish ending across a newline can never
|
||
# be valid JSON without a comma, so inserting one is always safe.
|
||
if idx + 1 < len(segs) and segs[idx + 1][0]:
|
||
body = s.rstrip()
|
||
tail_ws = s[len(body) :]
|
||
prev = body[-1:] if body else (fixed[-1].rstrip()[-1:] if fixed else "")
|
||
value_ending = prev in ('"', "}", "]") or prev.isdigit()
|
||
if "\n" in tail_ws and value_ending:
|
||
notes.append("inserted missing commas")
|
||
s = body + "," + tail_ws
|
||
fixed.append(s)
|
||
|
||
out = "".join(fixed)
|
||
# Trailing commas (safe now: strings are intact, commas here are structural)
|
||
segs2 = _tokenize(out, [])
|
||
parts: list[str] = []
|
||
changed = False
|
||
for is_str, seg in segs2:
|
||
if is_str:
|
||
parts.append(seg)
|
||
else:
|
||
new = re.sub(r",(\s*[}\]])", r"\1", seg)
|
||
changed = changed or new != seg
|
||
parts.append(new)
|
||
if changed:
|
||
notes.append("removed trailing commas")
|
||
return "".join(parts)
|
||
|
||
|
||
def _extract_and_balance(text: str, notes: list[str]) -> str:
|
||
"""Slice out the JSON object (dropping surrounding prose) and close any
|
||
unclosed braces/brackets."""
|
||
start = text.find("{")
|
||
if start == -1:
|
||
return text
|
||
if text[:start].strip():
|
||
notes.append("ignored text before the JSON block")
|
||
|
||
stack: list[str] = []
|
||
i, n = start, len(text)
|
||
end = -1
|
||
while i < n:
|
||
ch = text[i]
|
||
if ch == '"': # skip strings
|
||
i += 1
|
||
while i < n:
|
||
if text[i] == "\\":
|
||
i += 2
|
||
continue
|
||
if text[i] == '"':
|
||
break
|
||
i += 1
|
||
elif ch in "{[":
|
||
stack.append("}" if ch == "{" else "]")
|
||
elif ch in "}]":
|
||
if stack and stack[-1] == ch:
|
||
stack.pop()
|
||
if not stack:
|
||
end = i + 1
|
||
break
|
||
i += 1
|
||
|
||
if end != -1:
|
||
if text[end:].strip():
|
||
notes.append("ignored text after the JSON block")
|
||
return text[start:end]
|
||
# Ran out of input with open scopes: close them.
|
||
if stack:
|
||
notes.append("closed unclosed braces/brackets")
|
||
return text[start:] + "".join(reversed(stack))
|
||
return text[start:]
|
||
|
||
|
||
def repair_json_text(text: str) -> tuple[str, list[str]]:
|
||
"""
|
||
Best-effort repair of an almost-JSON snippet. Returns (candidate, notes)
|
||
where notes is a human-readable list of the fixes applied (deduplicated,
|
||
in order). Does NOT guarantee the result parses — callers still try
|
||
json.loads and surface its error if repair wasn't enough.
|
||
"""
|
||
notes: list[str] = []
|
||
t = _strip_fences(text, notes)
|
||
t = _normalize_unicode(t, notes)
|
||
t = t.strip()
|
||
|
||
# Brace-less inner fragment: "name": { ... } (with no outer braces).
|
||
# A quoted key at the start is a strong signal; a bare word only counts
|
||
# when followed by '{' so prose like "Note: ..." isn't swallowed.
|
||
if re.match(r'^\s*"[^"\n]+"\s*:\s*[{["]', t) or re.match(r"^\s*[A-Za-z_$][\w$.-]*\s*:\s*\{", t):
|
||
notes.append("wrapped fragment in braces")
|
||
t = "{" + t + "}"
|
||
|
||
t = _extract_and_balance(t, notes)
|
||
segs = _tokenize(t, notes)
|
||
t = _repair_segments(segs, notes)
|
||
|
||
# Re-balance in case comment/quote fixes exposed structure.
|
||
t = _extract_and_balance(t, [])
|
||
|
||
seen: set[str] = set()
|
||
unique = [x for x in notes if not (x in seen or seen.add(x))]
|
||
return t, unique
|
||
|
||
|
||
def parse_pasted_json_verbose(text: str) -> tuple[dict[str, dict], list[str]]:
|
||
"""
|
||
Like parse_pasted_json, but forgiving. Tries strict JSON first; if that
|
||
fails, runs repair_json_text() and retries. Returns (servers, repair_notes).
|
||
repair_notes is empty when the input was already valid.
|
||
"""
|
||
raw = (text or "").strip()
|
||
if not raw:
|
||
raise ValueError("Nothing to parse.")
|
||
try:
|
||
return _shape_servers(json.loads(raw)), []
|
||
except json.JSONDecodeError as strict_err:
|
||
candidate, notes = repair_json_text(raw)
|
||
try:
|
||
obj = json.loads(candidate)
|
||
except json.JSONDecodeError:
|
||
# Repair wasn't enough — report the original, more meaningful error.
|
||
raise ValueError(
|
||
f"Couldn't parse that as JSON even after auto-repair "
|
||
f"(line {strict_err.lineno}, column {strict_err.colno}: {strict_err.msg})."
|
||
) from strict_err
|
||
return _shape_servers(obj), notes
|
||
|
||
|
||
def _shape_servers(obj) -> dict[str, dict]:
|
||
"""Shared shape-detection for parsed paste content."""
|
||
if not isinstance(obj, dict):
|
||
raise ValueError("Top-level JSON must be an object ({ ... }).")
|
||
|
||
if isinstance(obj.get("mcpServers"), dict):
|
||
servers = obj["mcpServers"]
|
||
elif looks_like_server(obj):
|
||
servers = {suggest_name(obj): obj}
|
||
elif obj and all(isinstance(v, dict) for v in obj.values()):
|
||
servers = obj
|
||
else:
|
||
raise ValueError("Couldn't find any MCP server definitions in that JSON.")
|
||
|
||
clean: dict[str, dict] = {}
|
||
for name, data in servers.items():
|
||
if not looks_like_server(data):
|
||
raise ValueError(
|
||
f"'{name}' doesn't look like an MCP server (it needs a 'command' or a 'url')."
|
||
)
|
||
clean[str(name)] = data
|
||
return clean
|
||
|
||
|
||
def parse_pasted_json(text: str) -> dict[str, dict]:
|
||
"""
|
||
Accept any of the shapes MCP docs hand out and return {name: server_dict}:
|
||
|
||
1. Full config: {"mcpServers": {"name": {...}}}
|
||
2. Inner block only: {"name": {"command": ...}}
|
||
3. A bare server obj: {"command": ..., "args": [...]} (name is suggested)
|
||
|
||
Input doesn't have to be valid JSON — markdown fences, comments, trailing
|
||
commas, smart/single quotes, unquoted keys, surrounding prose, and missing
|
||
braces are repaired automatically (see repair_json_text).
|
||
|
||
Raises ValueError with a human message on anything unrecognizable.
|
||
"""
|
||
servers, _notes = parse_pasted_json_verbose(text)
|
||
return servers
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Secrets: detection + redaction
|
||
# --------------------------------------------------------------------------- #
|
||
_SECRET_KEY_RE = re.compile(
|
||
r"(?i)(token|secret|passw|api[-_]?key|apikey|auth|credential|bearer|private[-_]?key|access[-_]?key)"
|
||
)
|
||
|
||
# Well-known token prefixes that identify a bare value as a secret even
|
||
# without a telling key/flag name next to it.
|
||
_TOKEN_PREFIXES = (
|
||
"ghp_",
|
||
"gho_",
|
||
"ghu_",
|
||
"ghs_",
|
||
"github_pat_", # GitHub
|
||
"sk-",
|
||
"sk_live_",
|
||
"sk_test_",
|
||
"rk_live_", # OpenAI / Stripe
|
||
"xoxb-",
|
||
"xoxp-",
|
||
"xoxc-",
|
||
"xoxs-",
|
||
"xapp-", # Slack
|
||
"glpat-",
|
||
"gldt-", # GitLab
|
||
"AKIA",
|
||
"ASIA", # AWS access key IDs
|
||
"ya29.",
|
||
"AIza", # Google
|
||
"pypi-",
|
||
"npm_",
|
||
"dop_v1_",
|
||
"figd_",
|
||
)
|
||
|
||
MASK = "••••••••"
|
||
|
||
# Matches userinfo credentials embedded in a URL: scheme://user:pass@host
|
||
# Fires on postgres://user:pass@host but NOT on https://host/path or ssh://user@host.
|
||
_EMBEDDED_CRED_RE = re.compile(r"://[^:@/\s]+:[^:@/\s]+@")
|
||
|
||
|
||
def is_secret_key(name: str) -> bool:
|
||
"""Does this env-var / header / flag name look like it holds a secret?"""
|
||
return bool(_SECRET_KEY_RE.search(name or ""))
|
||
|
||
|
||
def _is_secret_value(value: str) -> bool:
|
||
return isinstance(value, str) and value.startswith(_TOKEN_PREFIXES)
|
||
|
||
|
||
def redact_args(args: list[str]) -> list[str]:
|
||
"""
|
||
Mask secret values in an args list for display/diagnostics:
|
||
--token abc123 -> --token •••••••• (value after a secret flag)
|
||
--api-key=abc123 -> --api-key=•••••••• (inline flag=value)
|
||
ghp_abc123 -> •••••••• (well-known token prefix)
|
||
Everything else passes through untouched.
|
||
"""
|
||
out: list[str] = []
|
||
mask_next = False
|
||
for a in args:
|
||
s = str(a)
|
||
if mask_next:
|
||
out.append(MASK)
|
||
mask_next = False
|
||
continue
|
||
if s.startswith("-") and "=" in s and is_secret_key(s.split("=", 1)[0]):
|
||
out.append(s.split("=", 1)[0] + "=" + MASK)
|
||
continue
|
||
if s.startswith("-") and is_secret_key(s):
|
||
out.append(s)
|
||
mask_next = True
|
||
continue
|
||
if _is_secret_value(s):
|
||
out.append(MASK)
|
||
continue
|
||
out.append(s)
|
||
return out
|
||
|
||
|
||
def args_secret_warning(data: dict) -> str | None:
|
||
"""
|
||
Return a warning string when any arg looks like a raw secret that would
|
||
be better placed in `env`. Returns None when no concern is found.
|
||
|
||
Skips --flag=value inline pairs (already partially self-documenting).
|
||
Fires on:
|
||
- positional values that start with a well-known token prefix (ghp_, sk-, …)
|
||
- values that follow a secret-named flag (--token abc, --api-key abc)
|
||
- URLs with embedded user:pass credentials (postgres://user:pass@host)
|
||
"""
|
||
args = [str(a) for a in (data.get("args") or [])]
|
||
mask_next = False
|
||
for a in args:
|
||
if mask_next:
|
||
mask_next = False
|
||
if not a.startswith("-"):
|
||
return (
|
||
"An arg value following a secret-named flag looks like a credential. "
|
||
"Where the server supports it, prefer Environment variables — "
|
||
"args are visible in process listings."
|
||
)
|
||
continue
|
||
# --flag=value inline: skip (the flag name already labels it)
|
||
if a.startswith("-") and "=" in a:
|
||
continue
|
||
# --secretflag (no inline value): flag the next positional arg
|
||
if a.startswith("-") and is_secret_key(a):
|
||
mask_next = True
|
||
continue
|
||
if a.startswith("-"):
|
||
continue
|
||
# Positional value: check for token prefix or embedded URL credentials
|
||
if _is_secret_value(a) or _EMBEDDED_CRED_RE.search(a):
|
||
return (
|
||
"An arg value looks like a credential. "
|
||
"Where the server supports it, prefer Environment variables — "
|
||
"args are visible in process listings."
|
||
)
|
||
return None
|
||
|
||
|
||
def split_suspicious_args(args: list[str]) -> tuple[list[str], list[str]]:
|
||
"""
|
||
Detect the classic argument-entry mistake: several argv tokens typed on one
|
||
line ("--directory /path/to/server"). An entry is only flagged when it
|
||
contains whitespace AND at least one whitespace-separated token starts with
|
||
"-" — so legitimate single arguments with spaces ("My Project Notes",
|
||
"/Users/me/My Documents") are never touched.
|
||
|
||
Returns (fixed_args, notes). notes is empty when nothing was suspicious;
|
||
otherwise it describes each split so a UI can show the proposed fix.
|
||
Quotes are respected when splitting ('--name "My Server"' becomes
|
||
['--name', 'My Server']).
|
||
"""
|
||
import shlex
|
||
|
||
out: list[str] = []
|
||
notes: list[str] = []
|
||
for a in args:
|
||
if isinstance(a, str) and _looks_like_multiple_args(a):
|
||
try:
|
||
parts = shlex.split(a)
|
||
except ValueError: # unbalanced quotes — fall back to plain split
|
||
parts = a.split()
|
||
if len(parts) > 1:
|
||
out.extend(parts)
|
||
notes.append(f"“{a}” looks like {len(parts)} arguments — split onto separate lines")
|
||
continue
|
||
out.append(a)
|
||
return out, notes
|
||
|
||
|
||
def _looks_like_multiple_args(a: str) -> bool:
|
||
toks = a.split()
|
||
return len(toks) > 1 and any(t.startswith("-") for t in toks)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Validation
|
||
# --------------------------------------------------------------------------- #
|
||
def validate_servers(servers: list[ServerEntry]) -> list[str]:
|
||
"""Return a list of human-readable problems. Empty list == all good."""
|
||
problems: list[str] = []
|
||
seen: dict[str, int] = {}
|
||
for s in servers:
|
||
nm = s.name.strip()
|
||
if not nm:
|
||
problems.append("A server has an empty name.")
|
||
seen[nm] = seen.get(nm, 0) + 1
|
||
if s.kind == "stdio":
|
||
if not str(s.data.get("command", "")).strip():
|
||
problems.append(f"'{nm or '(unnamed)'}' is a local server but has no command.")
|
||
else:
|
||
url = str(s.data.get("url", "")).strip()
|
||
if not url:
|
||
problems.append(f"'{nm or '(unnamed)'}' is a remote server but has no URL.")
|
||
elif not (url.startswith("http://") or url.startswith("https://")):
|
||
problems.append(f"'{nm}' has a URL that isn't http(s).")
|
||
for nm, count in seen.items():
|
||
if nm and count > 1:
|
||
problems.append(f"Duplicate server name: '{nm}' ({count}x).")
|
||
return problems
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Search / filter
|
||
# --------------------------------------------------------------------------- #
|
||
def server_matches_filter(entry: ServerEntry, query: str) -> bool:
|
||
"""
|
||
Case-insensitive substring match against a server's name, and its
|
||
command (stdio) or url (remote). An empty/whitespace-only query matches
|
||
everything -- that's what lets the search box double as "no filter".
|
||
"""
|
||
q = (query or "").strip().lower()
|
||
if not q:
|
||
return True
|
||
if q in entry.name.lower():
|
||
return True
|
||
if entry.kind == "remote":
|
||
haystack = str(entry.data.get("url", ""))
|
||
else:
|
||
haystack = str(entry.data.get("command", ""))
|
||
return q in haystack.lower()
|
||
|
||
|
||
def filter_servers(entries: list[ServerEntry], query: str) -> list[ServerEntry]:
|
||
"""Return only the entries that match `query` (see server_matches_filter)."""
|
||
return [e for e in entries if server_matches_filter(e, query)]
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 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)
|
||
|
||
|
||
def server_log_path(name: str) -> Path | None:
|
||
"""
|
||
The platform-specific Claude Desktop MCP server log file for `name`, or
|
||
None if it doesn't exist yet (nothing has been logged for this server).
|
||
|
||
macOS : ~/Library/Logs/Claude/mcp-server-<name>.log (one file per server)
|
||
Windows: %APPDATA%\\Claude\\logs\\mcp.log (one shared file)
|
||
Other platforms: Claude Desktop doesn't ship a log in a known location -> None.
|
||
"""
|
||
if sys.platform == "darwin":
|
||
p = Path.home() / "Library" / "Logs" / "Claude" / f"mcp-server-{name}.log"
|
||
elif sys.platform.startswith("win"):
|
||
appdata = Path(os.environ.get("APPDATA", Path.home() / "AppData" / "Roaming"))
|
||
p = appdata / "Claude" / "logs" / "mcp.log"
|
||
else:
|
||
return None
|
||
return p if p.is_file() else None
|
||
|
||
|
||
_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}",
|
||
}
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Health status (maps a spawn_test() result to a simple tri-state for the
|
||
# server-list UI's per-row status dot; see "Test all" in bcc.py)
|
||
# --------------------------------------------------------------------------- #
|
||
class HealthStatus:
|
||
"""
|
||
Tri-state health for the server-list status dot. A plain class of string
|
||
constants -- not an Enum -- to match the plain-string status values used
|
||
elsewhere in this module (see check_dependency's 'status').
|
||
"""
|
||
|
||
UNTESTED = "untested"
|
||
OK = "ok"
|
||
FAILED = "failed"
|
||
|
||
|
||
def health_from_spawn_result(result: dict) -> tuple[str, str]:
|
||
"""
|
||
Map a spawn_test() result dict to (HealthStatus, short_summary) for the
|
||
server-list status column. Reuses spawn_test's own outcome classification
|
||
rather than re-deriving pass/fail from returncode/stderr:
|
||
|
||
outcome "ok" -> OK (server started and kept running)
|
||
outcome "not_applicable" -> UNTESTED (remote server, or no command set)
|
||
anything else -> FAILED (exited, crashed, or not found)
|
||
|
||
The summary is short enough for a table cell/tooltip; when the process
|
||
wrote to stderr before dying, its first line is appended for context.
|
||
"""
|
||
outcome = result.get("outcome", "")
|
||
detail = result.get("detail", "") or ""
|
||
stderr = (result.get("stderr") or "").strip()
|
||
|
||
if outcome == "ok":
|
||
return HealthStatus.OK, detail or "started"
|
||
if outcome == "not_applicable":
|
||
return HealthStatus.UNTESTED, detail or "not applicable"
|
||
|
||
# exited / crashed / not_found: the server didn't come up cleanly.
|
||
summary = detail or outcome
|
||
if stderr:
|
||
first_line = stderr.splitlines()[0].strip()
|
||
if first_line:
|
||
summary = f"{summary} — {first_line}"
|
||
return HealthStatus.FAILED, summary
|
||
|
||
|
||
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
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Restart Claude Desktop (issue #9)
|
||
#
|
||
# Scoped strictly to Claude DESKTOP, the GUI app -- never Claude Code (the
|
||
# CLI), which has no long-running process to bounce. Callers should gate this
|
||
# behind profile_targets_claude_desktop() before offering it in the UI.
|
||
# --------------------------------------------------------------------------- #
|
||
class RestartResult(NamedTuple):
|
||
"""Outcome of a restart_claude_desktop() attempt."""
|
||
|
||
success: bool
|
||
detail: str
|
||
|
||
|
||
def _run_quiet(cmd: list[str]) -> None:
|
||
"""Best-effort fire-and-forget command. Never raises: a nonzero exit (e.g.
|
||
pkill finding nothing to kill) is expected and not an error."""
|
||
with contextlib.suppress(OSError):
|
||
subprocess.run(cmd, capture_output=True)
|
||
|
||
|
||
def restart_supported() -> bool:
|
||
"""
|
||
True only where restarting Claude Desktop makes sense (macOS, Windows).
|
||
There is no official Claude Desktop for Linux, and the obvious binary
|
||
name there ("claude") is the Claude Code CLI — killing or spawning it
|
||
would be actively harmful. GUI callers gate the Restart button on this.
|
||
"""
|
||
return sys.platform == "darwin" or sys.platform.startswith("win")
|
||
|
||
|
||
_MACOS_QUIT_WAIT_S = 5.0
|
||
|
||
|
||
def _macos_claude_running() -> bool:
|
||
try:
|
||
return subprocess.run(["pgrep", "-x", "Claude"], capture_output=True).returncode == 0
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _restart_claude_desktop_macos() -> RestartResult:
|
||
_run_quiet(["pkill", "-x", "Claude"])
|
||
# Wait for the old instance to actually exit: `open -a` against a dying
|
||
# process can merely re-activate it, and the config is only re-read on a
|
||
# true relaunch. Blocks up to _MACOS_QUIT_WAIT_S — callers run this off
|
||
# the UI thread (see RestartWorker in bcc.py).
|
||
deadline = time.monotonic() + _MACOS_QUIT_WAIT_S
|
||
while _macos_claude_running():
|
||
if time.monotonic() > deadline:
|
||
return RestartResult(
|
||
False,
|
||
f"Claude Desktop didn't quit within {_MACOS_QUIT_WAIT_S:.0f}s — "
|
||
"quit it manually, then reopen it.",
|
||
)
|
||
time.sleep(0.15)
|
||
try:
|
||
result = subprocess.run(["open", "-a", "Claude"], capture_output=True, text=True)
|
||
except OSError as e:
|
||
return RestartResult(False, f"Couldn't launch Claude Desktop: {e}")
|
||
if result.returncode != 0:
|
||
detail = (result.stderr or result.stdout or "").strip() or "'open -a Claude' failed"
|
||
return RestartResult(False, detail)
|
||
return RestartResult(True, "Claude Desktop restarted.")
|
||
|
||
|
||
def _claude_windows_start_menu_shortcut() -> Path:
|
||
appdata = os.environ.get("APPDATA", str(Path.home()))
|
||
return Path(appdata) / "Microsoft" / "Windows" / "Start Menu" / "Programs" / "Claude.lnk"
|
||
|
||
|
||
def _restart_claude_desktop_windows() -> RestartResult:
|
||
shortcut = _claude_windows_start_menu_shortcut()
|
||
if not shortcut.is_file():
|
||
# Checked BEFORE killing: an MSIX/Store install has no Start-menu .lnk
|
||
# at this path, and killing without a relaunch path would leave the
|
||
# user with no running Claude at all.
|
||
return RestartResult(
|
||
False,
|
||
f"Claude's Start-menu shortcut wasn't found ({shortcut}). "
|
||
"If Claude Desktop is installed from the Microsoft Store, "
|
||
"quit and reopen it manually.",
|
||
)
|
||
_run_quiet(["taskkill", "/IM", "Claude.exe", "/F"])
|
||
try:
|
||
# `cmd /c start "" <target>` launches detached, the same as double-clicking
|
||
# the Start-menu shortcut, and returns immediately.
|
||
result = subprocess.run(
|
||
["cmd", "/c", "start", "", str(shortcut)], capture_output=True, text=True
|
||
)
|
||
except OSError as e:
|
||
return RestartResult(False, f"Couldn't launch Claude Desktop: {e}")
|
||
if result.returncode != 0:
|
||
detail = (
|
||
result.stderr or result.stdout or ""
|
||
).strip() or "failed to relaunch Claude Desktop"
|
||
return RestartResult(False, detail)
|
||
return RestartResult(True, "Claude Desktop restarted.")
|
||
|
||
|
||
def restart_claude_desktop() -> RestartResult:
|
||
"""
|
||
Kill and relaunch the Claude Desktop app so a freshly saved config takes
|
||
effect. The app not currently running is NOT a failure -- pkill/taskkill
|
||
exiting non-zero just means "nothing to kill", and we go straight to
|
||
relaunching. Only a failed relaunch is reported as success=False.
|
||
|
||
macOS blocks for up to _MACOS_QUIT_WAIT_S while the old instance exits —
|
||
run off the UI thread. Unsupported platforms (see restart_supported())
|
||
refuse without touching any process.
|
||
"""
|
||
if sys.platform == "darwin":
|
||
return _restart_claude_desktop_macos()
|
||
if sys.platform.startswith("win"):
|
||
return _restart_claude_desktop_windows()
|
||
return RestartResult(False, "Restarting Claude Desktop isn't supported on this platform.")
|