ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(hermes_cli): mcp_catalog — split _parse_manifest into per-section parsers, compact install path

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:24:57 -07:00
parent e0e1ce28a2
commit eab02b0653
1 changed files with 174 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/mcp_catalog.py
+174 -304
View File
@@ -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.<name>.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_<NAME>_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_<NAME>_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/<name>`` and run
bootstrap commands. Returns the install directory."""
"""Clone the entry's repo into ``~/.hermes/mcp-installs/<name>`` 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.<name>`` 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.<key>`` (``include``/``exclude``) for *name*, if well-formed.
"""The user's prior ``tools.<key>`` (``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 <name>``.
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 <name>` 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 <name>` 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.<name>`` 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.<name>``; 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 <provider>` 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 <provider>` 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