From eab02b065365b31364e03af45dfcbfe3a0287c4e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:24:57 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20mcp=5Fcatalog=20?= =?UTF-8?q?=E2=80=94=20split=20=5Fparse=5Fmanifest=20into=20per-section=20?= =?UTF-8?q?parsers,=20compact=20install=20path?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/mcp_catalog.py | 478 ++++++++++++++------------------------ 1 file changed, 174 insertions(+), 304 deletions(-) diff --git a/hermes_cli/mcp_catalog.py b/hermes_cli/mcp_catalog.py index cf9f2a6062..c6413109ac 100644 --- a/hermes_cli/mcp_catalog.py +++ b/hermes_cli/mcp_catalog.py @@ -1,8 +1,7 @@ """MCP catalog — curated, Nous-approved MCP servers shipped with the repo. -Catalog policy: - Entries are added only by merging a PR into hermes-agent. Presence in the -``optional-mcps/`` directory = Nous approval. No community tier, no trust signals beyond "it's in -the catalog". - Manifests pin transport details (commands, args, refs). +Entries are added only by merging a PR into hermes-agent; presence in ``optional-mcps/`` = Nous +approval (no community tier, no other trust signals). Manifests pin transport details. """ from __future__ import annotations @@ -19,12 +18,7 @@ import yaml from hermes_constants import get_hermes_home, get_optional_mcps_dir from hermes_cli._subprocess_compat import noninteractive_git_env from hermes_cli.colors import Colors, color -from hermes_cli.config import ( - load_config, - save_config, - get_env_value, - save_env_value, -) +from hermes_cli.config import load_config, save_config, get_env_value, save_env_value from hermes_cli.cli_output import prompt as _prompt_input _MANIFEST_VERSION = 1 @@ -49,7 +43,7 @@ class EnvVarSpec: class AuthSpec: type: str # "api_key" | "oauth" | "none" env: List[EnvVarSpec] = field(default_factory=list) - # OAuth-specific (case 2: third-party provider like Google) + # OAuth-specific (third-party provider like Google) provider: Optional[str] = None scopes: List[str] = field(default_factory=list) env_var: Optional[str] = None @@ -62,9 +56,8 @@ class TransportSpec: args: List[str] = field(default_factory=list) url: Optional[str] = None version: Optional[str] = None # informational, pinned - # Static environment variables for the stdio subprocess (e.g. telemetry - # opt-outs, mode flags). NOT for secrets — credentials go through - # auth.env so they are prompted for and land in ~/.hermes/.env. + # Static env for the stdio subprocess (telemetry opt-outs, mode flags). NOT for secrets — + # credentials go through auth.env so they are prompted for and land in ~/.hermes/.env. env: Dict[str, str] = field(default_factory=dict) @@ -79,23 +72,15 @@ class InstallSpec: @dataclass class ToolsSpec: - """Manifest-side tool-selection hints. + """Manifest-side tool-selection hints: pre-check state for the install checklist and the + fallback selection when the probe fails (see install_entry()).""" - Drives the pre-checked state of the install-time tool checklist, and acts as the fallback - selection when probe fails. See install_entry() flow. - """ - - # If declared, these tool names are pre-checked in the checklist (or - # applied directly when probe fails). If None, all probed tools are - # pre-checked (or no filter is written when probe fails). + # Pre-checked (or applied directly on probe failure). None => all probed tools pre-checked + # (or no filter written on probe failure). default_enabled: Optional[List[str]] = None - - # Exclude-mode counterpart: tool names/glob patterns written to - # ``mcp_servers..tools.exclude`` at install time. Everything NOT - # matching stays enabled — including tools the server adds later. Use for - # huge auto-generated surfaces (OpenAPI-derived MCPs) where an include - # list would be thousands of lines and freeze out new endpoints. - # Mutually exclusive with ``default_enabled``. + # Exclude-mode counterpart written to ``tools.exclude``: everything NOT matching stays enabled, + # including tools the server adds later. For huge auto-generated surfaces (OpenAPI-derived + # MCPs). Mutually exclusive with ``default_enabled``. default_excluded: Optional[List[str]] = None @@ -103,15 +88,12 @@ class ToolsSpec: class SuggestSpec: """Composer-suggestion metadata (desktop "brand pill" triggers). - NOTE: GitHub is intentionally NOT in the catalog and must not be suggested here: its hosted MCP - requires a per-host OAuth app (generic DCR 404s), and the bundled github/* skills (gh CLI) are - the far more capable integration. Point users at the skills instead. + GitHub is intentionally NOT in the catalog and must not be suggested here: its hosted MCP needs a + per-host OAuth app (generic DCR 404s) and the bundled github/* skills are far more capable. """ - # Lowercase whole-word/phrase triggers matched against the draft. - keywords: List[str] = field(default_factory=list) - # Hostname suffixes ("atlassian.net") matched against pasted links. - hosts: List[str] = field(default_factory=list) + keywords: List[str] = field(default_factory=list) # lowercase whole-word/phrase triggers + hosts: List[str] = field(default_factory=list) # hostname suffixes ("atlassian.net") @dataclass @@ -136,9 +118,7 @@ class CatalogError(Exception): def _catalog_root() -> Path: - """Return the optional-mcps/ directory shipped with this Hermes install.""" - # Prefer the env-var override / packaged location; fall back to the repo's - # optional-mcps/ next to the package (source checkout). + """The optional-mcps/ dir: env-var override / packaged location, else the source checkout's.""" return get_optional_mcps_dir(Path(__file__).parent.parent / "optional-mcps") @@ -149,11 +129,8 @@ def _parse_env_spec(raw: Any) -> EnvVarSpec: if not name or not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name): raise CatalogError(f"invalid env var name: {name!r}") return EnvVarSpec( - name=name, - prompt=raw.get("prompt") or name, - required=bool(raw.get("required", True)), - secret=bool(raw.get("secret", True)), - default=str(raw.get("default") or ""), + name=name, prompt=raw.get("prompt") or name, required=bool(raw.get("required", True)), + secret=bool(raw.get("secret", True)), default=str(raw.get("default") or ""), ) @@ -170,14 +147,99 @@ def _require_list(path: Path, field: str, raw: Any) -> list: def _require_str_list(path: Path, field: str, raw: Any, *, non_empty: bool = False) -> None: - ok = isinstance(raw, list) and all( - isinstance(t, str) and (t.strip() if non_empty else True) for t in raw - ) + ok = isinstance(raw, list) and all(isinstance(t, str) and (t.strip() if non_empty else True) for t in raw) if not ok: kind = "non-empty strings" if non_empty else "strings" raise CatalogError(f"{path}: {field} must be a list of {kind}") +def _parse_transport(path: Path, raw: Any) -> TransportSpec: + transport_raw = _require_mapping(path, "transport", raw or {}) + t_type = transport_raw.get("type") + if t_type not in ("stdio", "http"): + raise CatalogError(f"{path}: transport.type must be 'stdio' or 'http'") + args = _require_list(path, "transport.args", transport_raw.get("args") or []) + env_raw = transport_raw.get("env") or {} + if not isinstance(env_raw, dict) or not all(isinstance(k, str) and isinstance(v, str) for k, v in env_raw.items()): + raise CatalogError(f"{path}: transport.env must be a mapping of string to string") + transport = TransportSpec( + type=t_type, command=transport_raw.get("command"), args=[str(a) for a in args], + url=transport_raw.get("url"), version=transport_raw.get("version"), env=dict(env_raw), + ) + if t_type == "stdio" and not transport.command: + raise CatalogError(f"{path}: stdio transport requires 'command'") + if t_type == "http" and not transport.url: + raise CatalogError(f"{path}: http transport requires 'url'") + return transport + + +def _parse_auth(path: Path, raw: Any, name: str, http: bool) -> AuthSpec: + auth_raw = _require_mapping(path, "auth", raw or {"type": "none"}) + a_type = auth_raw.get("type") or "none" + if a_type not in ("api_key", "oauth", "none"): + raise CatalogError(f"{path}: auth.type must be 'api_key'|'oauth'|'none'") + env_list = [_parse_env_spec(e) for e in _require_list(path, "auth.env", auth_raw.get("env") or [])] + if http and a_type == "api_key": + # _build_server_config emits an Authorization header referencing ${MCP__API_KEY}, but + # install_entry only persists the env vars DECLARED in auth.env. Enforce the naming contract + # here, or a manifest declaring e.g. N8N_API_KEY would send a literal-placeholder header (401). + from hermes_cli.mcp_config import _env_key_for_server + + _required_key = _env_key_for_server(name) + if all(spec.name != _required_key for spec in env_list): + raise CatalogError( + f"{path}: http + api_key auth requires auth.env to declare " + f"'{_required_key}' (the key the Authorization header references)" + ) + return AuthSpec( + type=a_type, env=env_list, provider=auth_raw.get("provider"), + scopes=list(auth_raw.get("scopes") or []), env_var=auth_raw.get("env_var"), + ) + + +def _parse_tools(path: Path, raw: Any) -> ToolsSpec: + tools_raw = _require_mapping(path, "tools", raw or {}) + default_enabled = tools_raw.get("default_enabled") + default_excluded = tools_raw.get("default_excluded") + for key, val in (("default_enabled", default_enabled), ("default_excluded", default_excluded)): + if val is not None: + _require_str_list(path, f"tools.{key}", val) + if default_enabled is not None and default_excluded is not None: + raise CatalogError(f"{path}: tools.default_enabled and tools.default_excluded are mutually exclusive") + return ToolsSpec(default_enabled=default_enabled, default_excluded=default_excluded) + + +def _parse_suggest(path: Path, suggest_raw: Any) -> Optional[SuggestSpec]: + if suggest_raw is None: + return None + _require_mapping(path, "suggest", suggest_raw) + kw_raw = suggest_raw.get("keywords") or [] + hosts_raw = suggest_raw.get("hosts") or [] + _require_str_list(path, "suggest.keywords", kw_raw, non_empty=True) + _require_str_list(path, "suggest.hosts", hosts_raw, non_empty=True) + if not kw_raw and not hosts_raw: + raise CatalogError(f"{path}: 'suggest' requires at least one keyword or host") + # Matching is case-insensitive whole-word / host-suffix: store lowercase so UIs needn't re-normalize. + return SuggestSpec( + keywords=[k.strip().lower() for k in kw_raw], + hosts=[h.strip().lower().lstrip(".") for h in hosts_raw], + ) + + +def _parse_install(path: Path, install_raw: Any) -> Optional[InstallSpec]: + if install_raw is None: + return None + _require_mapping(path, "install", install_raw) + i_type = install_raw.get("type") + if i_type != "git": + raise CatalogError(f"{path}: install.type must be 'git' (got {i_type!r})") + url, ref = install_raw.get("url") or "", install_raw.get("ref") or "" + if not url or not ref: + raise CatalogError(f"{path}: install.url and install.ref are required") + bootstrap = _require_list(path, "install.bootstrap", install_raw.get("bootstrap") or []) + return InstallSpec(type=i_type, url=url, ref=ref, bootstrap=[str(c) for c in bootstrap]) + + def _parse_manifest(path: Path) -> CatalogEntry: """Read and validate a manifest.yaml. Raise CatalogError on any problem.""" try: @@ -185,7 +247,6 @@ def _parse_manifest(path: Path) -> CatalogEntry: data = yaml.safe_load(f) or {} except Exception as exc: raise CatalogError(f"failed to read {path}: {exc}") from exc - if not isinstance(data, dict): raise CatalogError(f"{path}: manifest must be a mapping") @@ -195,125 +256,26 @@ def _parse_manifest(path: Path) -> CatalogEntry: f"{path}: manifest_version {mv!r} unsupported " f"(this Hermes understands version {_MANIFEST_VERSION})" ) - name = data.get("name") or "" if not name or not re.match(r"^[A-Za-z0-9_-]+$", name): raise CatalogError(f"{path}: invalid or missing 'name'") - description = str(data.get("description") or "").strip() if not description: raise CatalogError(f"{path}: 'description' required") - source = str(data.get("source") or "").strip() - - transport_raw = _require_mapping(path, "transport", data.get("transport") or {}) - t_type = transport_raw.get("type") - if t_type not in ("stdio", "http"): - raise CatalogError(f"{path}: transport.type must be 'stdio' or 'http'") - args = _require_list(path, "transport.args", transport_raw.get("args") or []) - env_raw = transport_raw.get("env") or {} - if not isinstance(env_raw, dict) or not all( - isinstance(k, str) and isinstance(v, str) for k, v in env_raw.items() - ): - raise CatalogError( - f"{path}: transport.env must be a mapping of string to string" - ) - transport = TransportSpec( - type=t_type, - command=transport_raw.get("command"), - args=[str(a) for a in args], - url=transport_raw.get("url"), - version=transport_raw.get("version"), - env=dict(env_raw), - ) - if t_type == "stdio" and not transport.command: - raise CatalogError(f"{path}: stdio transport requires 'command'") - if t_type == "http" and not transport.url: - raise CatalogError(f"{path}: http transport requires 'url'") - - auth_raw = _require_mapping(path, "auth", data.get("auth") or {"type": "none"}) - a_type = auth_raw.get("type") or "none" - if a_type not in ("api_key", "oauth", "none"): - raise CatalogError(f"{path}: auth.type must be 'api_key'|'oauth'|'none'") - env_list = [ - _parse_env_spec(e) - for e in _require_list(path, "auth.env", auth_raw.get("env") or []) - ] - auth = AuthSpec( - type=a_type, - env=env_list, - provider=auth_raw.get("provider"), - scopes=list(auth_raw.get("scopes") or []), - env_var=auth_raw.get("env_var"), - ) - if t_type == "http" and a_type == "api_key": - # _build_server_config emits an Authorization header referencing - # ${MCP__API_KEY} (via _bearer_auth_headers), but install_entry - # only persists the env vars DECLARED in auth.env. Enforce the naming - # contract at parse time, or a manifest declaring e.g. N8N_API_KEY - # would install cleanly yet send a literal-placeholder header (401) - # at connect time. - from hermes_cli.mcp_config import _env_key_for_server - - _required_key = _env_key_for_server(name) - if all(spec.name != _required_key for spec in env_list): - raise CatalogError( - f"{path}: http + api_key auth requires auth.env to declare " - f"'{_required_key}' (the key the Authorization header references)" - ) - - tools_raw = _require_mapping(path, "tools", data.get("tools") or {}) - default_enabled = tools_raw.get("default_enabled") - default_excluded = tools_raw.get("default_excluded") - for key, val in (("default_enabled", default_enabled), ("default_excluded", default_excluded)): - if val is not None: - _require_str_list(path, f"tools.{key}", val) - if default_enabled is not None and default_excluded is not None: - raise CatalogError( - f"{path}: tools.default_enabled and tools.default_excluded are " - "mutually exclusive" - ) - tools_spec = ToolsSpec(default_enabled=default_enabled, default_excluded=default_excluded) - - suggest: Optional[SuggestSpec] = None - suggest_raw = data.get("suggest") - if suggest_raw is not None: - _require_mapping(path, "suggest", suggest_raw) - kw_raw = suggest_raw.get("keywords") or [] - hosts_raw = suggest_raw.get("hosts") or [] - _require_str_list(path, "suggest.keywords", kw_raw, non_empty=True) - _require_str_list(path, "suggest.hosts", hosts_raw, non_empty=True) - if not kw_raw and not hosts_raw: - raise CatalogError( - f"{path}: 'suggest' requires at least one keyword or host" - ) - # Normalize: matching is case-insensitive whole-word / host-suffix, - # so store lowercase and let UIs match without re-normalizing. - suggest = SuggestSpec( - keywords=[k.strip().lower() for k in kw_raw], - hosts=[h.strip().lower().lstrip(".") for h in hosts_raw], - ) - - install: Optional[InstallSpec] = None - install_raw = data.get("install") - if install_raw is not None: - _require_mapping(path, "install", install_raw) - i_type = install_raw.get("type") - if i_type != "git": - raise CatalogError(f"{path}: install.type must be 'git' (got {i_type!r})") - url, ref = install_raw.get("url") or "", install_raw.get("ref") or "" - if not url or not ref: - raise CatalogError(f"{path}: install.url and install.ref are required") - bootstrap = _require_list(path, "install.bootstrap", install_raw.get("bootstrap") or []) - install = InstallSpec(type=i_type, url=url, ref=ref, bootstrap=[str(c) for c in bootstrap]) - + # Validation order (transport, auth, tools, suggest, install) determines which error surfaces. + transport = _parse_transport(path, data.get("transport")) + auth = _parse_auth(path, data.get("auth"), name, transport.type == "http") + tools = _parse_tools(path, data.get("tools")) + suggest = _parse_suggest(path, data.get("suggest")) + install = _parse_install(path, data.get("install")) return CatalogEntry( name=name, description=description, - source=source, + source=str(data.get("source") or "").strip(), transport=transport, auth=auth, - tools=tools_spec, + tools=tools, install=install, post_install=str(data.get("post_install") or ""), suggest=suggest, @@ -321,12 +283,16 @@ def _parse_manifest(path: Path) -> CatalogEntry: ) +# Populated by list_catalog(); inspected by the picker / catalog UIs so the user gets actionable +# feedback instead of a silently-shorter list. +_CATALOG_DIAGNOSTICS: List[tuple] = [] + + def list_catalog() -> List[CatalogEntry]: """Return all valid catalog entries, sorted by name. - Invalid manifests are skipped silently (CI catches them). Manifests with a future - ``manifest_version`` are also skipped but surfaced via :func:`catalog_diagnostics` so UIs can - tell the user their Hermes is out of date. + Invalid manifests are skipped silently (CI catches them); future ``manifest_version`` ones are + skipped too but surfaced via :func:`catalog_diagnostics` so UIs can say "update Hermes". """ root = _catalog_root() if not root.exists(): @@ -341,26 +307,14 @@ def list_catalog() -> List[CatalogEntry]: entries.append(_parse_manifest(manifest)) except CatalogError as exc: msg = str(exc) - # Recognize the future-manifest error specifically so the UI can - # surface a more actionable nudge than "broken manifest". future = "manifest_version" in msg and "unsupported" in msg - _CATALOG_DIAGNOSTICS.append( - (child.name, "future_manifest" if future else "invalid", msg) - ) + _CATALOG_DIAGNOSTICS.append((child.name, "future_manifest" if future else "invalid", msg)) return entries -# Populated by list_catalog(). Inspected by the picker / catalog UIs so the -# user gets actionable feedback instead of a silently-shorter list. -_CATALOG_DIAGNOSTICS: List[tuple] = [] - - def catalog_diagnostics() -> List[tuple]: - """Diagnostics from the most recent :func:`list_catalog` call. - - Returns ``(entry_name, kind, message)`` tuples; ``kind`` is ``future_manifest`` (newer than - this Hermes understands, update to install) or ``invalid`` (malformed, e.g. user-edited). - """ + """``(entry_name, kind, message)`` tuples from the most recent :func:`list_catalog` call; + ``kind`` is ``future_manifest`` (newer than this Hermes) or ``invalid`` (malformed).""" return list(_CATALOG_DIAGNOSTICS) @@ -376,8 +330,7 @@ def get_entry(name: str) -> Optional[CatalogEntry]: def installed_servers() -> Dict[str, dict]: """Return current ``mcp_servers`` block from config.yaml.""" - cfg = load_config() - servers = cfg.get("mcp_servers") or {} + servers = load_config().get("mcp_servers") or {} return servers if isinstance(servers, dict) else {} @@ -437,8 +390,7 @@ def _run_bootstrap(cwd: Path, commands: List[str]) -> None: def _do_git_install(entry: CatalogEntry) -> Path: - """Clone the entry's repo into ``~/.hermes/mcp-installs/`` and run - bootstrap commands. Returns the install directory.""" + """Clone the entry's repo into ``~/.hermes/mcp-installs/`` and run bootstrap. Returns the dir.""" assert entry.install is not None and entry.install.type == "git" install = entry.install dest = _install_root() / entry.name @@ -446,40 +398,26 @@ def _do_git_install(entry: CatalogEntry) -> Path: git = shutil.which("git") if not git: raise CatalogError("git is required to install this MCP but was not found on PATH") - if dest.exists(): - # Fresh checkout each install — manifest version is the source of truth, - # so wipe + re-clone for determinism. + # Fresh checkout each install — the manifest ref is the source of truth. _say(f" Removing existing install at {dest}", Colors.DIM) shutil.rmtree(dest) - _say(f" Cloning {install.url} ({install.ref}) → {dest}", Colors.CYAN) - # `git clone --branch` only accepts branches and tags, NOT commit SHAs. - # Detecting SHA-shaped refs upfront avoids a guaranteed stderr leak on - # the fast path (the --branch attempt would always fail noisily for a - # SHA ref before we fall back to full-clone-then-checkout). + # `git clone --branch` only accepts branches/tags, NOT commit SHAs; detect SHA-shaped refs + # upfront so the fast path doesn't always fail noisily before the full-clone fallback. is_sha_ref = bool(re.fullmatch(r"[0-9a-f]{7,40}", install.ref)) - - # Never let an install hang on a credential prompt: catalog installs run - # from CLI commands and dashboard flows where nobody can answer git's - # username/password prompt (private repo, bad remote, auth required). + # Never hang on a credential prompt: installs run from CLI/dashboard flows nobody can answer. _git_env = noninteractive_git_env() def _git(*args: str) -> int: - return subprocess.run( - [git, *args], stdin=subprocess.DEVNULL, env=_git_env - ).returncode + return subprocess.run([git, *args], stdin=subprocess.DEVNULL, env=_git_env).returncode - if not is_sha_ref and _git( - "clone", "--depth", "1", "--branch", install.ref, install.url, str(dest) - ) != 0: - # Branch/tag form failed (unlikely for valid manifests; possible if - # the ref was deleted upstream). Fall through to the full-clone path. + if not is_sha_ref and _git("clone", "--depth", "1", "--branch", install.ref, install.url, str(dest)) != 0: + # Branch/tag form failed (e.g. ref deleted upstream): fall through to full-clone path. if dest.exists(): shutil.rmtree(dest) - is_sha_ref = True # treat the same as a SHA ref from here - + is_sha_ref = True if is_sha_ref: if _git("clone", install.url, str(dest)) != 0: raise CatalogError(f"git clone failed for {install.url}") @@ -488,7 +426,6 @@ def _do_git_install(entry: CatalogEntry) -> Path: if install.bootstrap: _run_bootstrap(dest, install.bootstrap) - return dest @@ -496,15 +433,12 @@ def _expand_install_dir(value: str, install_dir: Optional[Path]) -> str: if _INSTALL_DIR_VAR not in value: return value if install_dir is None: - raise CatalogError( - f"manifest references {_INSTALL_DIR_VAR} but no install block exists" - ) + raise CatalogError(f"manifest references {_INSTALL_DIR_VAR} but no install block exists") return value.replace(_INSTALL_DIR_VAR, str(install_dir)) def _prompt_env_vars(specs: List[EnvVarSpec]) -> Dict[str, str]: - """Walk the env spec list, prompting the user for each. Writes secrets and - non-secrets alike to ~/.hermes/.env via save_env_value().""" + """Prompt for each env spec; secrets and non-secrets alike go to ~/.hermes/.env.""" collected: Dict[str, str] = {} for spec in specs: existing = get_env_value(spec.name) @@ -512,11 +446,7 @@ def _prompt_env_vars(specs: List[EnvVarSpec]) -> Dict[str, str]: _say(f" ✓ {spec.name} already set in .env") collected[spec.name] = existing continue - value = _prompt_input( - spec.prompt, - default=spec.default or None, - password=spec.secret, - ) + value = _prompt_input(spec.prompt, default=spec.default or None, password=spec.secret) if value: save_env_value(spec.name, value) collected[spec.name] = value @@ -525,9 +455,7 @@ def _prompt_env_vars(specs: List[EnvVarSpec]) -> Dict[str, str]: return collected -def _build_server_config( - entry: CatalogEntry, install_dir: Optional[Path] -) -> dict: +def _build_server_config(entry: CatalogEntry, install_dir: Optional[Path]) -> dict: """Translate a manifest into the ``mcp_servers.`` block format used by hermes_cli/mcp_config.py.""" cfg: dict = {} t = entry.transport @@ -549,11 +477,10 @@ def _build_server_config( def _read_prior_tool_list(name: str, key: str) -> Optional[List[str]]: - """Return the user's prior ``tools.`` (``include``/``exclude``) for *name*, if well-formed. + """The user's prior ``tools.`` (``include``/``exclude``) for *name*, if well-formed. - Read BEFORE a reinstall overwrites the server entry: a prior include list pre-checks the - checklist (tools no longer on the server are dropped at display time), and a user-edited - exclude list survives reinstall instead of being clobbered by the manifest's ``default_excluded``. + Read BEFORE a reinstall overwrites the entry: a prior include list pre-checks the checklist and a + user-edited exclude list survives instead of being clobbered by the manifest's ``default_excluded``. """ tools_cfg = (installed_servers().get(name) or {}).get("tools") or {} if not isinstance(tools_cfg, dict): @@ -566,21 +493,18 @@ def _read_prior_tool_list(name: str, key: str) -> Optional[List[str]]: def _probe_tools(name: str) -> Optional[List[tuple]]: """Connect to a freshly-configured MCP and list its tools. - Returns a list of ``(tool_name, description)`` tuples on success, or ``None`` on any failure - (server unreachable, OAuth not yet completed, backing service offline, etc.). Failures are - intentionally swallowed here — the fallback path in :func:`_apply_tool_selection` handles them. + ``(tool_name, description)`` tuples on success, ``None`` on any failure (unreachable, OAuth not + yet completed, ...). Failures are swallowed here; :func:`_apply_tool_selection` handles them. """ server_cfg = installed_servers().get(name) if not server_cfg: return None try: - # Import lazily so the catalog module stays cheap to load. - from hermes_cli.mcp_config import _probe_single_server + from hermes_cli.mcp_config import _probe_single_server # lazy: keep this module cheap tools = _probe_single_server(name, server_cfg) return list(tools) if tools is not None else [] except Exception as exc: - # Display the cause but never raise from the install path. _say(f" Probe failed: {exc}", Colors.YELLOW) return None @@ -613,32 +537,19 @@ def _apply_tool_selection( ) -> None: """Probe the server and let the user pick which tools to enable. - Probe-success path: - Curses checklist of all probed tools. - Pre-check uses (in priority - order): 1. *prior_selection* (reinstall: preserve what the user had) 2. manifest's - ``tools.default_enabled`` 3. all tools (default) - All-on selection clears any filter (no - ``tools.include`` written). - - Probe-fail path: - If manifest declares ``tools.default_enabled`` → apply directly. - Otherwise - → leave config with no filter (all on when reachable). - Either way, point the user at ``hermes - mcp configure ``. + Probe-success: curses checklist; pre-check priority *prior_selection* (reinstall) > manifest + ``tools.default_enabled`` > all; all-on clears any filter. Probe-fail: keep the prior filter, + else apply ``default_enabled``, else no filter; point the user at ``hermes mcp configure``. """ print() name = entry.name configure_hint = f"`hermes mcp configure {name}`" - # Exclude-mode manifests short-circuit the checklist entirely: the curated - # exclude list (names or glob patterns) is written as-is, everything else - # stays enabled — including tools the server adds later. Reinstalls - # preserve the user's own prior filter in EITHER mode: a prior include - # selection falls through to the checklist below, and a prior user-edited - # exclude list is re-written verbatim instead of being clobbered by the - # manifest defaults. - # (No probe announcement here — this path deliberately never probes.) + # Exclude-mode manifests never probe: the curated exclude list (names or globs) is written as-is + # and everything else stays enabled, including tools the server adds later. A prior include + # selection falls through to the checklist; a prior user-edited exclude list is kept verbatim. if entry.tools.default_excluded and prior_selection is None: - edit_hint = ( - f"Edit mcp_servers.{name}.tools.exclude in config.yaml or run " - f"{configure_hint} to change." - ) + edit_hint = f"Edit mcp_servers.{name}.tools.exclude in config.yaml or run {configure_hint} to change." if prior_exclude is not None: _write_tools_filter(name, "exclude", prior_exclude) _say(f" Kept your existing exclude list ({len(prior_exclude)} entries). {edit_hint}") @@ -653,10 +564,8 @@ def _apply_tool_selection( _say(f" Probing '{name}' for available tools...", Colors.CYAN) probed = _probe_tools(name) - # Probe failure path. Order matters: a reinstall must come out of a - # failed probe with the user's previous filter intact (common for OAuth - # entries — the entry rewrite precedes first auth, so the server is - # regularly unreachable right here), not with the filter reset or wiped. + # Probe failure. Order matters: a reinstall must keep the user's previous filter intact (common + # for OAuth entries — the entry rewrite precedes first auth, so the server is unreachable here). if probed is None: manifest_default = entry.tools.default_enabled refine_hint = f"Run {configure_hint} after the server is reachable to refine." @@ -685,67 +594,44 @@ def _apply_tool_selection( return if not probed: - # Probe succeeded but server reported zero tools. Nothing to filter. _write_tools_filter(name, "include", None) _say(" Server reported no tools.", Colors.YELLOW) return tool_names = [t[0] for t in probed] - # Non-TTY: skip the checklist. Priority matches the interactive - # pre-check priority: prior user selection > manifest default > all-on. + # Non-TTY: skip the checklist; same priority as the interactive pre-check. import sys as _sys if not _sys.stdin.isatty(): - preferred = ( - prior_selection if prior_selection is not None - else (entry.tools.default_enabled or None) - ) - _write_tools_filter( - name, "include", None if preferred is None else [n for n in preferred if n in tool_names] - ) + preferred = prior_selection if prior_selection is not None else (entry.tools.default_enabled or None) + _write_tools_filter(name, "include", None if preferred is None else [n for n in preferred if n in tool_names]) return - # Build the pre-checked set in priority order - pre_set = { - n for n in (prior_selection or entry.tools.default_enabled or tool_names) - if n in tool_names - } + pre_set = {n for n in (prior_selection or entry.tools.default_enabled or tool_names) if n in tool_names} pre_indices = {i for i, n in enumerate(tool_names) if n in pre_set} - _say(f" Found {len(probed)} tool(s). Pre-checked: {len(pre_indices)}.") from hermes_cli.curses_ui import curses_checklist - labels = [ - f"{n} — {(d[:60] + '...') if len(d) > 60 else d}" - for n, d in probed - ] + labels = [f"{n} — {(d[:60] + '...') if len(d) > 60 else d}" for n, d in probed] chosen_indices = curses_checklist( - f"Select tools for '{name}' (SPACE toggle, ENTER confirm)", - labels, - pre_indices, + f"Select tools for '{name}' (SPACE toggle, ENTER confirm)", labels, pre_indices, ) - if not chosen_indices: - # User unchecked everything; treat as "no tools" — write empty include - # so the server is installed but contributes nothing until reconfigured. + # Everything unchecked: write an empty include so the server is installed but contributes + # nothing until reconfigured. _write_tools_filter(name, "include", []) _say(f" No tools selected. Run {configure_hint} to change.", Colors.YELLOW) return - if len(chosen_indices) == len(probed): - # Everything selected — clear filter for the cleanest config shape. - # NOTE: this means any tools the server adds later (e.g. a future MCP - # version) will also be auto-enabled. To pin to the current set, - # the user can re-run `hermes mcp configure ` and unselect a - # tool to switch back to include-mode. + # Clear the filter: tools the server adds later are auto-enabled too. To pin the current set, + # re-run `hermes mcp configure ` and unselect a tool (switches to include-mode). _write_tools_filter(name, "include", None) _say( f" ✓ All {len(probed)} tools enabled (no filter — new tools " "the server adds later will be auto-enabled)." ) return - chosen_names = [tool_names[i] for i in sorted(chosen_indices)] _write_tools_filter(name, "include", chosen_names) _say(f" ✓ {len(chosen_names)}/{len(probed)} tools enabled.") @@ -754,10 +640,9 @@ def _apply_tool_selection( def install_entry(entry: CatalogEntry, *, enable: bool = True) -> None: """Install a catalog entry end-to-end. - Order: git clone + bootstrap (if ``install.type == git``); API-key prompt to .env or the - ``auth: oauth`` marker; translate the manifest into ``mcp_servers.`` in config.yaml; - probe the server and offer a tool checklist (falling back to ``tools.default_enabled`` or - all-on when the probe fails); print post_install notes. + Order: git clone + bootstrap (if any); API-key prompt to .env or the ``auth: oauth`` marker; + write ``mcp_servers.``; probe + tool checklist (falling back per + :func:`_apply_tool_selection`); print post_install notes. """ print() _say(f" Installing MCP '{entry.name}'", Colors.CYAN + Colors.BOLD) @@ -769,16 +654,13 @@ def install_entry(entry: CatalogEntry, *, enable: bool = True) -> None: install_dir = _do_git_install(entry) if entry.install is not None else None - # Auth if entry.auth.type == "api_key": print() _say(" Configure credentials:", Colors.CYAN) _prompt_env_vars(entry.auth.env) elif entry.auth.type == "oauth" and entry.auth.provider: - # Case 2: provider-mediated (Google, GitHub, etc.). We rely on - # the existing `hermes auth ` flow. Surface guidance - # here rather than auto-running it — keeps the catalog install - # decoupled from provider-auth lifecycle. + # Provider-mediated OAuth relies on the existing `hermes auth ` flow; surface + # guidance rather than auto-running it to keep install decoupled from provider-auth lifecycle. _say( f" This MCP uses {entry.auth.provider} OAuth. Run " f"`hermes auth {entry.auth.provider}` if you have not " @@ -791,31 +673,20 @@ def install_entry(entry: CatalogEntry, *, enable: bool = True) -> None: "on first connection (browser flow).", Colors.DIM, ) - # auth.type == "none": nothing to do. - # ── Preserve any prior user tool selection across reinstalls ──────── - # Reading BEFORE we overwrite the entry below so a reinstall pre-checks - # whatever the user picked last time (include mode) or keeps the user's - # edited exclude list (exclude mode). + # Read prior user selection BEFORE overwriting the entry so a reinstall preserves it. prior_selection = _read_prior_tool_list(entry.name, "include") prior_exclude = _read_prior_tool_list(entry.name, "exclude") - # Build and write the mcp_servers entry (without tools filter yet; - # _apply_tool_selection() finalizes it below). server_cfg = _build_server_config(entry, install_dir) server_cfg["enabled"] = enable from hermes_cli.mcp_config import _save_mcp_server if not _save_mcp_server(entry.name, server_cfg): - raise CatalogError( - f"catalog entry '{entry.name}' rejected: suspicious command/args configuration" - ) + raise CatalogError(f"catalog entry '{entry.name}' rejected: suspicious command/args configuration") - # ── Probe + tool selection ────────────────────────────────────────── - _apply_tool_selection( - entry, prior_selection=prior_selection, prior_exclude=prior_exclude - ) + _apply_tool_selection(entry, prior_selection=prior_selection, prior_exclude=prior_exclude) print() _say( @@ -831,8 +702,7 @@ def install_entry(entry: CatalogEntry, *, enable: bool = True) -> None: def uninstall_entry(name: str, *, purge_install_dir: bool = True) -> bool: - """Remove a catalog-installed MCP from config and (optionally) wipe its - clone directory. Returns True if anything was removed.""" + """Remove a catalog-installed MCP from config and (optionally) its clone dir. True if anything was removed.""" removed = remove_server(name) if purge_install_dir: clone = _install_root() / name