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

refactor(hclib): remaining hermes_cli library modules — dead code, unified helpers, flattened branches

Browse Source
This commit is contained in:
Teknium
2026-09-02 14:38:33 -07:00
parent 78335adfec
commit cfa327e5dc
131 changed files with 10763 additions and 19250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/mcp_catalog.py
+234 -392
View File
@@ -1,28 +1,8 @@
"""MCP catalog — curated, Nous-approved MCP servers shipped with the repo.
Mirrors the optional-skills/ pattern: each catalog entry lives under
``optional-mcps/<name>/manifest.yaml`` and ships disabled. Users discover
entries via ``hermes mcp catalog`` or the interactive ``hermes mcp picker``,
and install them with ``hermes mcp install <name>`` (or by toggling in the
picker, which flows them through any required env/OAuth setup).
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). Pins follow the
same supply-chain rules as pyproject dependencies: exact versions for
package launchers (``uvx pkg==X``, ``npx pkg@X``), full commit SHAs for
git installs, and the pinned release should be at least 2 weeks old at
pin time. MCPs are never
auto-updated; users explicitly re-run ``hermes mcp install <name>`` to
pull a new manifest version after a repo update.
- Secrets prompted at install time go to ``~/.hermes/.env`` (the
.env-is-for-secrets rule). Non-secret env vars also go to .env to keep
one credential store.
See website/docs/user-guide/mcp-catalog.md for user docs.
See references/mcp-catalog.md (this repo's skill) for the manifest schema.
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).
"""
from __future__ import annotations
@@ -90,10 +70,7 @@ class TransportSpec:
@dataclass
class InstallSpec:
"""Optional bootstrap step (git clone + dep install).
Omit for one-shot launchable servers (npx, uvx).
"""
"""Optional bootstrap step (git clone + dep install)."""
type: str # "git"
url: str
ref: str # commit/tag/branch — pinned, never floats
@@ -104,8 +81,8 @@ class InstallSpec:
class ToolsSpec:
"""Manifest-side tool-selection hints.
Drives the pre-checked state of the install-time tool checklist, and acts
as the fallback selection when probe fails. See install_entry() flow.
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
@@ -126,16 +103,9 @@ class ToolsSpec:
class SuggestSpec:
"""Composer-suggestion metadata (desktop "brand pill" triggers).
Optional. When present, UI surfaces (currently the desktop composer)
may suggest installing this entry when the user's draft contains one
of the keywords as a completed whole word, or pastes a link whose
hostname ends with one of the host suffixes. Purely advisory — the
install itself always flows through the ordinary validated paths.
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.
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.
"""
# Lowercase whole-word/phrase triggers matched against the draft.
@@ -187,6 +157,27 @@ def _parse_env_spec(raw: Any) -> EnvVarSpec:
)
def _require_mapping(path: Path, key: str, raw: Any) -> dict:
if not isinstance(raw, dict):
raise CatalogError(f"{path}: '{key}' must be a mapping")
return raw
def _require_list(path: Path, field: str, raw: Any) -> list:
if not isinstance(raw, list):
raise CatalogError(f"{path}: {field} must be a list")
return raw
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
)
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_manifest(path: Path) -> CatalogEntry:
"""Read and validate a manifest.yaml. Raise CatalogError on any problem."""
try:
@@ -215,15 +206,11 @@ def _parse_manifest(path: Path) -> CatalogEntry:
source = str(data.get("source") or "").strip()
transport_raw = data.get("transport") or {}
if not isinstance(transport_raw, dict):
raise CatalogError(f"{path}: 'transport' must be a mapping")
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 = transport_raw.get("args") or []
if not isinstance(args, list):
raise CatalogError(f"{path}: transport.args must be a list")
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()
@@ -244,16 +231,14 @@ def _parse_manifest(path: Path) -> CatalogEntry:
if t_type == "http" and not transport.url:
raise CatalogError(f"{path}: http transport requires 'url'")
auth_raw = data.get("auth") or {"type": "none"}
if not isinstance(auth_raw, dict):
raise CatalogError(f"{path}: 'auth' must be a mapping")
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_raw = auth_raw.get("env") or []
if not isinstance(env_list_raw, list):
raise CatalogError(f"{path}: auth.env must be a list")
env_list = [_parse_env_spec(e) for e in env_list_raw]
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,
@@ -271,59 +256,33 @@ def _parse_manifest(path: Path) -> CatalogEntry:
from hermes_cli.mcp_config import _env_key_for_server
_required_key = _env_key_for_server(name)
if not any(spec.name == _required_key for spec in env_list):
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 = data.get("tools") or {}
if not isinstance(tools_raw, dict):
raise CatalogError(f"{path}: 'tools' must be a mapping")
tools_raw = _require_mapping(path, "tools", data.get("tools") or {})
default_enabled = tools_raw.get("default_enabled")
if default_enabled is not None:
if not isinstance(default_enabled, list) or not all(
isinstance(t, str) for t in default_enabled
):
raise CatalogError(
f"{path}: tools.default_enabled must be a list of strings"
)
default_excluded = tools_raw.get("default_excluded")
if default_excluded is not None:
if not isinstance(default_excluded, list) or not all(
isinstance(t, str) for t in default_excluded
):
raise CatalogError(
f"{path}: tools.default_excluded must be a list of strings"
)
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
)
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:
if not isinstance(suggest_raw, dict):
raise CatalogError(f"{path}: 'suggest' must be a mapping")
_require_mapping(path, "suggest", suggest_raw)
kw_raw = suggest_raw.get("keywords") or []
hosts_raw = suggest_raw.get("hosts") or []
if not isinstance(kw_raw, list) or not all(
isinstance(k, str) and k.strip() for k in kw_raw
):
raise CatalogError(
f"{path}: suggest.keywords must be a list of non-empty strings"
)
if not isinstance(hosts_raw, list) or not all(
isinstance(h, str) and h.strip() for h in hosts_raw
):
raise CatalogError(
f"{path}: suggest.hosts must be a list of non-empty strings"
)
_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"
@@ -338,24 +297,15 @@ def _parse_manifest(path: Path) -> CatalogEntry:
install: Optional[InstallSpec] = None
install_raw = data.get("install")
if install_raw is not None:
if not isinstance(install_raw, dict):
raise CatalogError(f"{path}: 'install' must be a mapping")
_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 = install_raw.get("url") or ""
ref = install_raw.get("ref") or ""
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 = install_raw.get("bootstrap") or []
if not isinstance(bootstrap, list):
raise CatalogError(f"{path}: install.bootstrap must be a list")
install = InstallSpec(
type=i_type,
url=url,
ref=ref,
bootstrap=[str(c) for c in bootstrap],
)
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])
return CatalogEntry(
name=name,
@@ -374,10 +324,9 @@ def _parse_manifest(path: Path) -> CatalogEntry:
def list_catalog() -> List[CatalogEntry]:
"""Return all valid catalog entries, sorted by name.
Invalid manifests are skipped silently (CI tests catch them at PR time).
Manifests with a future ``manifest_version`` are also skipped, but the
skip is surfaced via :func:`catalog_diagnostics` so the picker / catalog
UIs can tell the user their Hermes is out of date.
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.
"""
root = _catalog_root()
if not root.exists():
@@ -394,11 +343,10 @@ def list_catalog() -> List[CatalogEntry]:
msg = str(exc)
# Recognize the future-manifest error specifically so the UI can
# surface a more actionable nudge than "broken manifest".
if "manifest_version" in msg and "unsupported" in msg:
_CATALOG_DIAGNOSTICS.append((child.name, "future_manifest", msg))
else:
_CATALOG_DIAGNOSTICS.append((child.name, "invalid", msg))
continue
future = "manifest_version" in msg and "unsupported" in msg
_CATALOG_DIAGNOSTICS.append(
(child.name, "future_manifest" if future else "invalid", msg)
)
return entries
@@ -410,12 +358,8 @@ _CATALOG_DIAGNOSTICS: List[tuple] = []
def catalog_diagnostics() -> List[tuple]:
"""Diagnostics from the most recent :func:`list_catalog` call.
Returns a list of ``(entry_name, kind, message)`` tuples where ``kind``
is one of:
- ``future_manifest`` — manifest_version is newer than this Hermes
understands. Update Hermes to install this entry.
- ``invalid`` — manifest is malformed in some other way (caught by
CI for shipped manifests; user-modified manifests can hit this).
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).
"""
return list(_CATALOG_DIAGNOSTICS)
@@ -424,10 +368,7 @@ def get_entry(name: str) -> Optional[CatalogEntry]:
"""Look up a single entry by name. ``official/<name>`` prefix accepted."""
if name.startswith("official/"):
name = name[len("official/"):]
for entry in list_catalog():
if entry.name == name:
return entry
return None
return next((e for e in list_catalog() if e.name == name), None)
# ─── Status helpers ──────────────────────────────────────────────────────────
@@ -444,20 +385,41 @@ def is_installed(name: str) -> bool:
return name in installed_servers()
def is_enabled(name: str) -> bool:
servers = installed_servers()
cfg = servers.get(name)
if not cfg:
return False
def server_enabled(cfg: dict) -> bool:
"""Interpret a server block's ``enabled`` flag (bools, and yes/true/1 strings)."""
enabled = cfg.get("enabled", True)
if isinstance(enabled, str):
return enabled.lower() in {"true", "1", "yes"}
return bool(enabled)
def is_enabled(name: str) -> bool:
cfg = installed_servers().get(name)
return bool(cfg) and server_enabled(cfg)
def remove_server(name: str) -> bool:
"""Drop ``mcp_servers.<name>`` from config.yaml (pruning an empty block). True if it existed."""
cfg = load_config()
servers = cfg.get("mcp_servers") or {}
if name not in servers:
return False
del servers[name]
if not servers:
cfg.pop("mcp_servers", None)
else:
cfg["mcp_servers"] = servers
save_config(cfg)
return True
# ─── Install ─────────────────────────────────────────────────────────────────
def _say(msg: str, colour: str = Colors.GREEN) -> None:
print(color(msg, colour))
def _install_root() -> Path:
"""Where git-bootstrapped MCPs are cloned. Per-user, profile-aware."""
root = get_hermes_home() / "mcp-installs"
@@ -466,18 +428,12 @@ def _install_root() -> Path:
def _run_bootstrap(cwd: Path, commands: List[str]) -> None:
"""Execute bootstrap commands in *cwd*. Raise CatalogError on first failure.
Each command runs through the shell (so `&&` etc. work). The output is
streamed to the user's terminal for visibility.
"""
"""Execute bootstrap commands in *cwd*. Raise CatalogError on first failure."""
for cmd in commands:
print(color(f" $ {cmd}", Colors.DIM))
proc = subprocess.run(cmd, cwd=str(cwd), shell=True)
if proc.returncode != 0:
raise CatalogError(
f"bootstrap step failed (exit {proc.returncode}): {cmd}"
)
_say(f" $ {cmd}", Colors.DIM)
rc = subprocess.run(cmd, cwd=str(cwd), shell=True).returncode
if rc != 0:
raise CatalogError(f"bootstrap step failed (exit {rc}): {cmd}")
def _do_git_install(entry: CatalogEntry) -> Path:
@@ -494,10 +450,10 @@ def _do_git_install(entry: CatalogEntry) -> Path:
if dest.exists():
# Fresh checkout each install — manifest version is the source of truth,
# so wipe + re-clone for determinism.
print(color(f" Removing existing install at {dest}", Colors.DIM))
_say(f" Removing existing install at {dest}", Colors.DIM)
shutil.rmtree(dest)
print(color(f" Cloning {install.url} ({install.ref}) → {dest}", Colors.CYAN))
_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
@@ -510,35 +466,24 @@ def _do_git_install(entry: CatalogEntry) -> Path:
# username/password prompt (private repo, bad remote, auth required).
_git_env = noninteractive_git_env()
if not is_sha_ref:
proc = subprocess.run(
[git, "clone", "--depth", "1", "--branch", install.ref, install.url, str(dest)],
stdin=subprocess.DEVNULL,
env=_git_env,
)
if proc.returncode == 0:
pass
else:
# Branch/tag form failed (unlikely for valid manifests; possible if
# the ref was deleted upstream). Fall through to the full-clone path.
if dest.exists():
shutil.rmtree(dest)
is_sha_ref = True # treat the same as a SHA ref from here
def _git(*args: str) -> int:
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 dest.exists():
shutil.rmtree(dest)
is_sha_ref = True # treat the same as a SHA ref from here
if is_sha_ref:
proc = subprocess.run(
[git, "clone", install.url, str(dest)],
stdin=subprocess.DEVNULL,
env=_git_env,
)
if proc.returncode != 0:
if _git("clone", install.url, str(dest)) != 0:
raise CatalogError(f"git clone failed for {install.url}")
proc = subprocess.run(
[git, "-C", str(dest), "checkout", install.ref],
stdin=subprocess.DEVNULL,
env=_git_env,
)
if proc.returncode != 0:
if _git("-C", str(dest), "checkout", install.ref) != 0:
raise CatalogError(f"git checkout {install.ref} failed")
if install.bootstrap:
@@ -564,7 +509,7 @@ def _prompt_env_vars(specs: List[EnvVarSpec]) -> Dict[str, str]:
for spec in specs:
existing = get_env_value(spec.name)
if existing:
print(color(f" ✓ {spec.name} already set in .env", Colors.GREEN))
_say(f" ✓ {spec.name} already set in .env")
collected[spec.name] = existing
continue
value = _prompt_input(
@@ -572,20 +517,18 @@ def _prompt_env_vars(specs: List[EnvVarSpec]) -> Dict[str, str]:
default=spec.default or None,
password=spec.secret,
)
if not value:
if spec.required:
raise CatalogError(f"{spec.name} is required but no value was provided")
continue
save_env_value(spec.name, value)
collected[spec.name] = value
if value:
save_env_value(spec.name, value)
collected[spec.name] = value
elif spec.required:
raise CatalogError(f"{spec.name} is required but no value was provided")
return collected
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."""
"""Translate a manifest into the ``mcp_servers.<name>`` block format used by hermes_cli/mcp_config.py."""
cfg: dict = {}
t = entry.transport
if t.type == "stdio":
@@ -605,53 +548,29 @@ def _build_server_config(
return cfg
def _read_prior_tool_selection(name: str) -> Optional[List[str]]:
"""Return the user's prior `tools.include` for *name*, if any.
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.
Used during reinstalls so the install-time checklist starts pre-checked
with whatever the user already had. Tools no longer on the server are
silently dropped at checklist-display time.
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``.
"""
servers = installed_servers()
cfg = servers.get(name) or {}
tools_cfg = cfg.get("tools") or {}
tools_cfg = (installed_servers().get(name) or {}).get("tools") or {}
if not isinstance(tools_cfg, dict):
return None
include = tools_cfg.get("include")
if isinstance(include, list) and all(isinstance(t, str) for t in include):
return list(include)
return None
def _read_prior_tool_exclude(name: str) -> Optional[List[str]]:
"""Return the user's prior `tools.exclude` for *name*, if any.
The exclude-mode counterpart of :func:`_read_prior_tool_selection`.
Read BEFORE a reinstall overwrites the server entry, so a user-edited
exclude list survives reinstalling an exclude-mode catalog entry instead
of being clobbered by the manifest's ``default_excluded``.
"""
servers = installed_servers()
cfg = servers.get(name) or {}
tools_cfg = cfg.get("tools") or {}
if not isinstance(tools_cfg, dict):
return None
exclude = tools_cfg.get("exclude")
if isinstance(exclude, list) and all(isinstance(t, str) for t in exclude):
return list(exclude)
return None
value = tools_cfg.get(key)
ok = isinstance(value, list) and all(isinstance(t, str) for t in value)
return list(value) if ok else None
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.
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.
"""
servers = installed_servers()
server_cfg = servers.get(name)
server_cfg = installed_servers().get(name)
if not server_cfg:
return None
try:
@@ -662,46 +581,30 @@ def _probe_tools(name: str) -> Optional[List[tuple]]:
return list(tools) if tools is not None else []
except Exception as exc:
# Display the cause but never raise from the install path.
print(color(f" Probe failed: {exc}", Colors.YELLOW))
_say(f" Probe failed: {exc}", Colors.YELLOW)
return None
def _write_tools_include(name: str, include: Optional[List[str]]) -> None:
"""Persist or clear ``mcp_servers.<name>.tools.include``."""
def _write_tools_filter(name: str, mode: str, values: Optional[List[str]]) -> None:
"""Persist ``mcp_servers.<name>.tools.<mode>`` (``include``/``exclude``), clearing the other
mode; ``values=None`` drops the whole tools block (no filter)."""
cfg = load_config()
servers = cfg.setdefault("mcp_servers", {})
server_entry = servers.get(name) or {}
if include is None:
# No filter — drop any existing tools block.
if values is None:
server_entry.pop("tools", None)
else:
tools_block = server_entry.get("tools") or {}
if not isinstance(tools_block, dict):
tools_block = {}
tools_block["include"] = list(include)
tools_block.pop("exclude", None)
tools_block[mode] = list(values)
tools_block.pop("exclude" if mode == "include" else "include", None)
server_entry["tools"] = tools_block
servers[name] = server_entry
cfg["mcp_servers"] = servers
save_config(cfg)
def _write_tools_exclude(name: str, exclude: List[str]) -> None:
"""Persist ``mcp_servers.<name>.tools.exclude`` (names or glob patterns)."""
cfg = load_config()
servers = cfg.setdefault("mcp_servers", {})
server_entry = servers.get(name) or {}
tools_block = server_entry.get("tools") or {}
if not isinstance(tools_block, dict):
tools_block = {}
tools_block["exclude"] = list(exclude)
tools_block.pop("include", None)
server_entry["tools"] = tools_block
servers[name] = server_entry
cfg["mcp_servers"] = servers
save_config(cfg)
def _apply_tool_selection(
entry: CatalogEntry,
*,
@@ -710,21 +613,18 @@ 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).
- Sub-selection writes ``tools.include``.
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-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>``.
"""
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
@@ -735,29 +635,23 @@ def _apply_tool_selection(
# manifest defaults.
# (No probe announcement here — this path deliberately never probes.)
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."
)
if prior_exclude is not None:
_write_tools_exclude(entry.name, prior_exclude)
print(color(
f" Kept your existing exclude list ({len(prior_exclude)} "
f"entries). Edit mcp_servers.{entry.name}.tools.exclude in "
"config.yaml or run "
f"`hermes mcp configure {entry.name}` to change.",
Colors.GREEN,
))
_write_tools_filter(name, "exclude", prior_exclude)
_say(f" Kept your existing exclude list ({len(prior_exclude)} entries). {edit_hint}")
return
_write_tools_exclude(entry.name, entry.tools.default_excluded)
print(color(
f" Applied manifest exclude list "
f"({len(entry.tools.default_excluded)} entries); everything else "
f"stays enabled. Edit mcp_servers.{entry.name}.tools.exclude in "
"config.yaml or run "
f"`hermes mcp configure {entry.name}` to change.",
Colors.GREEN,
))
_write_tools_filter(name, "exclude", entry.tools.default_excluded)
_say(
f" Applied manifest exclude list ({len(entry.tools.default_excluded)} entries); "
f"everything else stays enabled. {edit_hint}"
)
return
print(color(f" Probing '{entry.name}' for available tools...", Colors.CYAN))
probed = _probe_tools(entry.name)
_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
@@ -765,79 +659,60 @@ def _apply_tool_selection(
# regularly unreachable right here), not with the filter reset or wiped.
if probed is None:
manifest_default = entry.tools.default_enabled
refine_hint = f"Run {configure_hint} after the server is reachable to refine."
if prior_selection is not None:
_write_tools_include(entry.name, prior_selection)
print(color(
f" Couldn\'t probe server. Kept your previous tool "
f"selection ({len(prior_selection)} tools). "
f"Run `hermes mcp configure {entry.name}` after the server "
"is reachable to refine.",
Colors.YELLOW,
))
_write_tools_filter(name, "include", prior_selection)
msg = (
f" Couldn't probe server. Kept your previous tool selection "
f"({len(prior_selection)} tools). {refine_hint}"
)
elif prior_exclude is not None:
_write_tools_exclude(entry.name, prior_exclude)
print(color(
f" Couldn\'t probe server. Kept your existing exclude "
f"list ({len(prior_exclude)} entries).",
Colors.YELLOW,
))
_write_tools_filter(name, "exclude", prior_exclude)
msg = f" Couldn't probe server. Kept your existing exclude list ({len(prior_exclude)} entries)."
elif manifest_default:
_write_tools_include(entry.name, manifest_default)
print(color(
f" Couldn\'t probe server. Applied manifest default "
f"({len(manifest_default)} tools). "
f"Run `hermes mcp configure {entry.name}` after the server "
"is reachable to refine.",
Colors.YELLOW,
))
_write_tools_filter(name, "include", manifest_default)
msg = (
f" Couldn't probe server. Applied manifest default "
f"({len(manifest_default)} tools). {refine_hint}"
)
else:
_write_tools_include(entry.name, None)
print(color(
f" Couldn\'t probe server; installed with no tool filter "
"(all tools enabled when reachable). "
f"Run `hermes mcp configure {entry.name}` after first "
"connect to prune.",
Colors.YELLOW,
))
_write_tools_filter(name, "include", None)
msg = (
" Couldn't probe server; installed with no tool filter (all tools enabled when "
f"reachable). Run {configure_hint} after first connect to prune."
)
_say(msg, Colors.YELLOW)
return
if not probed:
# Probe succeeded but server reported zero tools. Nothing to filter.
_write_tools_include(entry.name, None)
print(color(" Server reported no tools.", Colors.YELLOW))
_write_tools_filter(name, "include", None)
_say(" Server reported no tools.", Colors.YELLOW)
return
tool_names = [t[0] for t in probed]
# Build the pre-checked set in priority order
if prior_selection:
pre_set = {n for n in prior_selection if n in tool_names}
elif entry.tools.default_enabled:
pre_set = {n for n in entry.tools.default_enabled if n in tool_names}
else:
pre_set = set(tool_names)
pre_indices = {i for i, n in enumerate(tool_names) if n in pre_set}
# Non-TTY: skip the checklist. Priority matches the interactive
# pre-check priority: prior user selection > manifest default > all-on.
import sys as _sys
if not _sys.stdin.isatty():
if prior_selection is not None:
include = [n for n in prior_selection if n in tool_names]
_write_tools_include(entry.name, include)
elif entry.tools.default_enabled:
include = [n for n in entry.tools.default_enabled if n in tool_names]
_write_tools_include(entry.name, include)
else:
_write_tools_include(entry.name, None)
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
print(color(
f" Found {len(probed)} tool(s). "
f"Pre-checked: {len(pre_indices)}.",
Colors.GREEN,
))
# 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_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
@@ -846,7 +721,7 @@ def _apply_tool_selection(
for n, d in probed
]
chosen_indices = curses_checklist(
f"Select tools for '{entry.name}' (SPACE toggle, ENTER confirm)",
f"Select tools for '{name}' (SPACE toggle, ENTER confirm)",
labels,
pre_indices,
)
@@ -854,12 +729,8 @@ def _apply_tool_selection(
if not chosen_indices:
# User unchecked everything; treat as "no tools" — write empty include
# so the server is installed but contributes nothing until reconfigured.
_write_tools_include(entry.name, [])
print(color(
f" No tools selected. Run `hermes mcp configure {entry.name}` "
"to change.",
Colors.YELLOW,
))
_write_tools_filter(name, "include", [])
_say(f" No tools selected. Run {configure_hint} to change.", Colors.YELLOW)
return
if len(chosen_indices) == len(probed):
@@ -868,82 +739,66 @@ def _apply_tool_selection(
# 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.
_write_tools_include(entry.name, None)
print(color(
_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).",
Colors.GREEN,
))
"the server adds later will be auto-enabled)."
)
return
chosen_names = [tool_names[i] for i in sorted(chosen_indices)]
_write_tools_include(entry.name, chosen_names)
print(color(
f" ✓ {len(chosen_names)}/{len(probed)} tools enabled.",
Colors.GREEN,
))
_write_tools_filter(name, "include", chosen_names)
_say(f" ✓ {len(chosen_names)}/{len(probed)} tools enabled.")
def install_entry(entry: CatalogEntry, *, enable: bool = True) -> None:
"""Install a catalog entry end-to-end.
Steps:
1. If ``install.type == git``, clone + run bootstrap commands.
2. If ``auth.type == api_key``, prompt for env vars, save to .env.
3. If ``auth.type == oauth`` (remote MCP / case 1), write the
``auth: oauth`` marker (MCP client handles browser on first connect
in the non-pre-authenticated case).
4. Translate the manifest into an ``mcp_servers.<name>`` block and
save into config.yaml.
5. Probe the server, present a curses checklist for tool selection,
write ``tools.include`` (or no filter, depending on choice).
If probe fails, fall back to the manifest's
``tools.default_enabled`` or all-on.
6. Print post_install notes.
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.
"""
print()
print(color(f" Installing MCP '{entry.name}'", Colors.CYAN + Colors.BOLD))
_say(f" Installing MCP '{entry.name}'", Colors.CYAN + Colors.BOLD)
if entry.description:
print(color(f" {entry.description}", Colors.DIM))
_say(f" {entry.description}", Colors.DIM)
if entry.source:
print(color(f" Source: {entry.source}", Colors.DIM))
_say(f" Source: {entry.source}", Colors.DIM)
print()
install_dir: Optional[Path] = None
if entry.install is not None:
install_dir = _do_git_install(entry)
install_dir = _do_git_install(entry) if entry.install is not None else None
# Auth
if entry.auth.type == "api_key":
print()
print(color(" Configure credentials:", Colors.CYAN))
_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.
_say(
f" This MCP uses {entry.auth.provider} OAuth. Run "
f"`hermes auth {entry.auth.provider}` if you have not "
"already authenticated.",
Colors.YELLOW,
)
elif entry.auth.type == "oauth":
if 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.
print(color(
f" This MCP uses {entry.auth.provider} OAuth. Run "
f"`hermes auth {entry.auth.provider}` if you have not "
"already authenticated.",
Colors.YELLOW,
))
else:
print(color(
" This MCP uses native OAuth 2.1; tokens will be acquired "
"on first connection (browser flow).",
Colors.DIM,
))
_say(
" This MCP uses native OAuth 2.1; tokens will be acquired "
"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).
prior_selection = _read_prior_tool_selection(entry.name)
prior_exclude = _read_prior_tool_exclude(entry.name)
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).
@@ -963,38 +818,25 @@ def install_entry(entry: CatalogEntry, *, enable: bool = True) -> None:
)
print()
print(color(
_say(
f" ✓ Installed '{entry.name}' "
f"({'enabled' if enable else 'disabled'}). "
f"Start a new Hermes session to load its tools.",
Colors.GREEN,
))
f"Start a new Hermes session to load its tools."
)
if entry.post_install:
print()
for line in entry.post_install.strip().splitlines():
print(color(f" {line}", Colors.DIM))
_say(f" {line}", Colors.DIM)
print()
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."""
cfg = load_config()
servers = cfg.get("mcp_servers") or {}
removed = False
if name in servers:
del servers[name]
if not servers:
cfg.pop("mcp_servers", None)
else:
cfg["mcp_servers"] = servers
save_config(cfg)
removed = True
removed = remove_server(name)
if purge_install_dir:
clone = _install_root() / name
if clone.exists():
shutil.rmtree(clone)
removed = True
return removed