From e0e1ce28a25fc7b8bec542d350313d2f2d33944e Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:20:47 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20mcp=5Fconfig=20?= =?UTF-8?q?=E2=80=94=20extract=20add-flow=20auth/tool-choice=20and=20exclu?= =?UTF-8?q?de-rebuild=20helpers,=20shared=20tool=20printer,=20compact?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/mcp_config.py | 700 +++++++++++++-------------------------- 1 file changed, 234 insertions(+), 466 deletions(-) diff --git a/hermes_cli/mcp_config.py b/hermes_cli/mcp_config.py index df6069510f..c0119dbcd1 100644 --- a/hermes_cli/mcp_config.py +++ b/hermes_cli/mcp_config.py @@ -24,15 +24,10 @@ logger = logging.getLogger(__name__) _ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") - _MCP_PRESETS: Dict[str, Dict[str, Any]] = { - "codex": { - "command": "codex", - "args": ["mcp-server"], - }, + "codex": {"command": "codex", "args": ["mcp-server"]}, } - # ─── UI Helpers ─────────────────────────────────────────────────────────────── def _info(text: str): @@ -55,14 +50,13 @@ def _confirm(question: str, default: bool = True) -> bool: except (KeyboardInterrupt, EOFError): print() return default - if not val: - return default - return val in {"y", "yes"} + return val in {"y", "yes"} if val else default -def _prompt(question: str, *, password: bool = False, default: str = "") -> str: - from hermes_cli.cli_output import prompt as _shared_prompt - return _shared_prompt(question, default=default, password=password) +def _print_tools(tools: List[Tuple[str, str]], width: int, desc_max: int) -> None: + for tool_name, desc in tools: + short = desc[:desc_max] + "..." if len(desc) > desc_max else desc + print(f" {color(tool_name, Colors.GREEN):{width}s} {short}") # ─── Config Helpers ─────────────────────────────────────────────────────────── @@ -72,9 +66,7 @@ def _get_mcp_servers(config: Optional[dict] = None) -> Dict[str, dict]: if config is None: config = load_config() servers = config.get("mcp_servers") - if not servers or not isinstance(servers, dict): - return {} - return servers + return servers if servers and isinstance(servers, dict) else {} def _tool_filters(cfg: dict) -> Tuple[Optional[list], Optional[list]]: @@ -82,8 +74,7 @@ def _tool_filters(cfg: dict) -> Tuple[Optional[list], Optional[list]]: tools_cfg = cfg.get("tools", {}) if not isinstance(tools_cfg, dict): return None, None - include = tools_cfg.get("include") - exclude = tools_cfg.get("exclude") + include, exclude = tools_cfg.get("include"), tools_cfg.get("exclude") return ( include if include and isinstance(include, list) else None, exclude if exclude and isinstance(exclude, list) else None, @@ -93,9 +84,8 @@ def _tool_filters(cfg: dict) -> Tuple[Optional[list], Optional[list]]: def _save_mcp_server(name: str, server_config: dict) -> bool: """Add or update a server entry in config.yaml. - Returns False when a high-signal exfiltration-shaped stdio command is rejected. stdio servers - are user-chosen local commands, so this blocks shell+egress payloads rather than whitelisting - command families. + Returns False when a high-signal exfiltration-shaped stdio command is rejected (shell+egress + payloads are blocked rather than whitelisting command families). """ if not _validate_or_warn(name, server_config): return False @@ -141,9 +131,8 @@ def _remove_mcp_server(name: str) -> bool: def _replace_mcp_servers(servers: Dict[str, dict]) -> Tuple[bool, List[str]]: """Replace the WHOLE ``mcp_servers`` map in config.yaml. - Every entry is validated up front; on any suspicious command/args the whole save is rejected - (returns ``(False, issues)``) so a bad paste can't be partially applied. An empty map removes - the key entirely. + Every entry is validated up front; any suspicious entry rejects the whole save (``(False, + issues)``) so a bad paste can't be partially applied. An empty map removes the key entirely. """ issues: List[str] = [] for name, cfg in servers.items(): @@ -151,10 +140,8 @@ def _replace_mcp_servers(servers: Dict[str, dict]) -> Tuple[bool, List[str]]: issues.append(f"Server '{name}': expected an object") continue issues.extend(validate_mcp_server_entry(name, cfg)) - if issues: return False, issues - config = load_config() if servers: config["mcp_servers"] = dict(servers) @@ -174,7 +161,7 @@ def _strip_bearer_prefix(token: str) -> str: """Strip a leading ``Bearer `` from a pasted token. The header template already stores ``Authorization: Bearer ${MCP_X_API_KEY}``; a token pasted - with its own prefix would send ``Bearer Bearer `` and get a 401, so normalize on save. + with its own prefix would send ``Bearer Bearer `` and get a 401. """ if not isinstance(token, str): return token @@ -185,22 +172,16 @@ def _strip_bearer_prefix(token: str) -> str: def _bearer_auth_headers(name: str) -> Dict[str, str]: - """Build the persisted Authorization header for a named MCP server. + """Build the persisted Authorization header template for a named MCP server. - The secret itself lives in the active profile's ``.env`` file. Keeping this template - construction beside ``_env_key_for_server`` ensures the CLI and Dashboard produce byte- - equivalent MCP configuration. + The secret lives in the profile's ``.env``; CLI and Dashboard share this so they produce + byte-equivalent config. """ - env_key = _env_key_for_server(name) - return {"Authorization": f"Bearer ${{{env_key}}}"} + return {"Authorization": f"Bearer ${{{_env_key_for_server(name)}}}"} def _save_bearer_auth_token(name: str, token: str) -> Dict[str, str]: - """Persist a Bearer token in the active profile and return safe headers. - - ``token`` is a one-time provisioning value. It is normalized and written only to ``.env``; - callers persist the returned interpolation template in ``config.yaml``. - """ + """Persist a normalized Bearer token to ``.env`` and return the header template for config.yaml.""" normalized = _strip_bearer_prefix(token) if not normalized or normalized.lower() == "bearer": raise ValueError("Bearer token is required") @@ -239,25 +220,19 @@ def _apply_mcp_preset( """Apply a known MCP preset when transport details were omitted.""" if not preset_name: return url, command, cmd_args, False - preset = _MCP_PRESETS.get(preset_name) if not preset: raise ValueError(f"Unknown MCP preset: {preset_name}") - if url or command: return url, command, cmd_args, False - - url = preset.get("url") - command = preset.get("command") + url, command = preset.get("url"), preset.get("command") cmd_args = list(preset.get("args") or []) - if url: server_config["url"] = url if command: server_config["command"] = command if cmd_args: server_config["args"] = cmd_args - return url, command, cmd_args, True @@ -266,13 +241,10 @@ def _apply_mcp_preset( def _resolve_mcp_server_config(config: dict) -> dict: """Resolve ``${ENV}`` placeholders in a server config before connecting. - Mirrors ``_load_mcp_config()`` in ``tools/mcp_tool.py``: load ``~/.hermes/.env`` and interpolate - ``${VAR}`` recursively. Without this the discovery probe sent the literal placeholder in header - templates and auth-requiring servers returned 401 while runtime loading (which interpolates) - worked. + Mirrors ``_load_mcp_config()`` in ``tools/mcp_tool.py``; without it the discovery probe sent + literal placeholders in header templates and auth-requiring servers returned 401. """ from tools.mcp_tool import _interpolate_env_vars - from agent.secret_scope import current_secret_scope if current_secret_scope() is None: @@ -290,26 +262,20 @@ def _probe_single_server( """Temporarily connect to one MCP server, list its tools, disconnect. Returns ``(tool_name, description)`` tuples; raises on connection failure. ``details`` is an - out-param filled with extra capability counts (``prompts``, ``resources``) so the return shape - stays stable for existing callers. + out-param filled with ``schema_chars``/``prompts``/``resources`` so the return shape stays stable. """ issues = validate_mcp_server_entry(name, config) if issues: raise ValueError("; ".join(issues)) from tools.mcp_tool import ( - _ensure_mcp_loop, - _run_on_mcp_loop, - _connect_server, - _stop_mcp_loop_if_idle, - _parse_boolish, + _ensure_mcp_loop, _run_on_mcp_loop, _connect_server, _stop_mcp_loop_if_idle, _parse_boolish, ) config = _resolve_mcp_server_config(config) if connect_timeout is None: - raw_timeout = config.get("connect_timeout", 30) try: - connect_timeout = max(1.0, float(raw_timeout)) + connect_timeout = max(1.0, float(config.get("connect_timeout", 30))) except (TypeError, ValueError): connect_timeout = 30.0 @@ -317,81 +283,49 @@ def _probe_single_server( tools_found: List[Tuple[str, str]] = [] async def _probe(): - server = await asyncio.wait_for( - _connect_server(name, config), timeout=connect_timeout - ) + server = await asyncio.wait_for(_connect_server(name, config), timeout=connect_timeout) try: for t in server._tools: desc = getattr(t, "description", "") or "" - # Truncate long descriptions for display if len(desc) > 80: desc = desc[:77] + "..." tools_found.append((t.name, desc)) if details is not None: - # Per-tool registry-schema sizes so the desktop can estimate the - # per-call token cost a server adds. Uses the SAME converted - # schema the agent registers (name + description + normalized - # parameters) — i.e. what actually rides on every model call. - # Additive-optional wire field: best-effort, absent on failure. + # Per-tool registry-schema sizes (the SAME converted schema the agent registers) so + # the desktop can estimate per-call token cost. Best-effort, absent on failure. try: import json as _json - from tools.mcp_tool import _convert_mcp_schema details["schema_chars"] = { - t.name: len( - _json.dumps( - _convert_mcp_schema(name, t), - separators=(",", ":"), - default=str, - ) - ) + t.name: len(_json.dumps( + _convert_mcp_schema(name, t), separators=(",", ":"), default=str, + )) for t in server._tools } except Exception: # pragma: no cover — display-only extra pass - if details is not None: - # Gate the capability probes exactly like runtime utility-tool - # registration (tools.mcp_tool._select_utility_schemas): - # 1. honour the user's tools.prompts / tools.resources config - # 2. only call a family the server actually advertises. - # Without this the "Test server" probe fired prompts/list and - # resources/list at every server unconditionally — so a server - # that rejects those methods (e.g. Unreal's MCP server, which - # answers "Call to unknown method 'prompts/list'") logged a hard - # error, and setting tools.prompts: false did NOT suppress it. + # Gate capability probes like runtime registration (_select_utility_schemas): + # honour tools.prompts / tools.resources config AND only call a family the server + # advertises — some servers hard-error on unknown prompts/list. tools_filter = config.get("tools") or {} - prompts_enabled = _parse_boolish( - tools_filter.get("prompts"), default=True - ) - resources_enabled = _parse_boolish( - tools_filter.get("resources"), default=True - ) - advertised_caps = getattr( - getattr(server, "initialize_result", None), - "capabilities", - None, - ) + advertised_caps = getattr(getattr(server, "initialize_result", None), "capabilities", None) - def _advertises(cap_attr: str) -> bool: - # When no capability info was captured (legacy fixtures / - # older servers) preserve the old always-try behaviour. - if advertised_caps is None: - return True - return getattr(advertised_caps, cap_attr, None) is not None + def _wanted(cap: str) -> bool: + # No capability info captured (legacy fixtures / older servers) => always try. + if not _parse_boolish(tools_filter.get(cap), default=True): + return False + return advertised_caps is None or getattr(advertised_caps, cap, None) is not None - # Capability probes are best-effort: servers without the - # capability raise, which just means "0". - if prompts_enabled and _advertises("prompts"): + # Best-effort: servers without the capability raise, which just means "0". + if _wanted("prompts"): try: - result = await server.session.list_prompts() - details["prompts"] = len(result.prompts) + details["prompts"] = len((await server.session.list_prompts()).prompts) except Exception: pass - if resources_enabled and _advertises("resources"): + if _wanted("resources"): try: - result = await server.session.list_resources() - details["resources"] = len(result.resources) + details["resources"] = len((await server.session.list_resources()).resources) except Exception: pass finally: @@ -403,68 +337,113 @@ def _probe_single_server( raise _unwrap_exception_group(exc) from None finally: _stop_mcp_loop_if_idle() - return tools_found def _oauth_tokens_present(name: str) -> bool: - """Return True if an OAuth token file exists on disk for ``name``. - - Used after ``hermes mcp login`` to distinguish a genuine authentication from a probe that - succeeded only because the server allowed initialize/tools-list without auth (so no token was - ever acquired). - """ + """True if an OAuth token file exists for ``name`` (a clean probe alone is not proof of auth).""" try: from tools.mcp_oauth import HermesTokenStorage return HermesTokenStorage(name).has_cached_tokens() except Exception as exc: # pragma: no cover — defensive logger.debug("Could not check OAuth tokens for '%s': %s", name, exc) - # Be permissive on unexpected errors: don't block a real success. - return True + return True # permissive: don't block a real success def _unwrap_exception_group(exc: BaseException) -> Exception: - """Extract the root-cause exception from anyio TaskGroup wrappers. - - The MCP SDK's anyio task groups wrap errors in ``ExceptionGroup``, making messages opaque - ("unhandled errors in a TaskGroup"); unwrap to surface the real cause, e.g. "401 Unauthorized". - """ + """Extract the root cause from anyio ``ExceptionGroup`` wrappers so e.g. "401 Unauthorized" surfaces.""" while isinstance(exc, BaseExceptionGroup) and exc.exceptions: exc = exc.exceptions[0] - # Return a plain Exception so callers can catch normally - if isinstance(exc, Exception): - return exc - return RuntimeError(str(exc)) + return exc if isinstance(exc, Exception) else RuntimeError(str(exc)) # ─── hermes mcp add ────────────────────────────────────────────────────────── +def _configure_http_auth(name: str, url: str, auth_type: Optional[str], server_config: Dict[str, Any]) -> bool: + """OAuth or Bearer-token setup for an HTTP server. False when the user cancelled.""" + print() + if auth_type == "oauth": + _info(f"Starting OAuth flow for '{name}'...") + oauth_ok = False + try: + from tools.mcp_oauth_manager import get_manager + if get_manager().get_or_build_provider(name, url, server_config.get("oauth")): + server_config["auth"] = "oauth" + _success("OAuth configured (tokens will be acquired on first connection)") + oauth_ok = True + else: + _warning("OAuth setup failed — MCP SDK auth module not available") + except Exception as exc: + _warning(f"OAuth error: {exc}") + if not oauth_ok: + _info("This server may not support OAuth.") + if not _confirm("Continue without authentication?", default=True): + _info("Cancelled.") + return False + return True + + _info(f"Connecting to {url}") + if _confirm("Does this server require authentication?", default=True) and (auth_type == "header" or not auth_type): + env_key = _env_key_for_server(name) + if get_env_value(env_key): + _success(f"{env_key}: already configured") + server_config["headers"] = _bearer_auth_headers(name) + else: + from hermes_cli.cli_output import prompt + api_key = prompt("API key / Bearer token", default="", password=True) + if api_key: + server_config["headers"] = _save_bearer_auth_token(name, api_key) + _success(f"Saved to {display_hermes_home()}/.env as {env_key}") + return True + + +def _choose_tools(name: str, tools: List[Tuple[str, str]], server_config: Dict[str, Any]) -> Optional[int]: + """Ask enable-all / select / cancel; returns the enabled-tool count or None when cancelled.""" + print() + _success(f"Connected! Found {len(tools)} tool(s) from '{name}':") + print() + _print_tools(tools, 40, 60) + print() + try: + choice = input(color(f" Enable all {len(tools)} tools? [Y/n/select]: ", Colors.YELLOW)).strip().lower() + except (KeyboardInterrupt, EOFError): + print() + _info("Cancelled.") + return None + if choice in {"n", "no"}: + _info("Cancelled — server not saved.") + return None + if choice not in {"s", "select"}: + return len(tools) + from hermes_cli.curses_ui import curses_checklist + + chosen = curses_checklist(f"Select tools for '{name}'", [f"{t[0]} — {t[1]}" for t in tools], set(range(len(tools)))) + if not chosen: + _info("No tools selected — server not saved.") + return None + chosen_names = [tools[i][0] for i in sorted(chosen)] + server_config.setdefault("tools", {})["include"] = chosen_names + return len(chosen_names) + + def cmd_mcp_add(args): """Add a new MCP server with discovery-first tool selection.""" name = args.name url = getattr(args, "url", None) - # Read from `mcp_command` (set by --command via explicit dest) — see - # mcp_add_p.add_argument("--command", dest="mcp_command", ...) in - # hermes_cli/main.py for why the dest is renamed. + # --command uses dest="mcp_command" (see hermes_cli/main.py for why the dest is renamed). command = getattr(args, "mcp_command", None) cmd_args = getattr(args, "args", None) or [] if cmd_args and cmd_args[0] == "--": cmd_args = cmd_args[1:] auth_type = getattr(args, "auth", None) - preset_name = getattr(args, "preset", None) - raw_env = getattr(args, "env", None) raw_connect_timeout = getattr(args, "connect_timeout", None) server_config: Dict[str, Any] = {} try: - explicit_env = _parse_env_assignments(raw_env) + explicit_env = _parse_env_assignments(getattr(args, "env", None)) url, command, cmd_args, _preset_applied = _apply_mcp_preset( - name, - preset_name=preset_name, - url=url, - command=command, - cmd_args=list(cmd_args), - server_config=server_config, + name, preset_name=getattr(args, "preset", None), url=url, command=command, + cmd_args=list(cmd_args), server_config=server_config, ) except ValueError as exc: _error(str(exc)) @@ -473,8 +452,6 @@ def cmd_mcp_add(args): if url and explicit_env: _error("--env is only supported for stdio MCP servers (--command or stdio presets)") return - - # Validate transport if not url and not command: _error("Must specify --url , --command , or --preset ") _info("Examples:") @@ -483,14 +460,10 @@ def cmd_mcp_add(args): _info(' hermes mcp add myserver --preset mypreset') return - # Check if server already exists - existing = _get_mcp_servers() - if name in existing: - if not _confirm(f"Server '{name}' already exists. Overwrite?", default=False): - _info("Cancelled.") - return + if name in _get_mcp_servers() and not _confirm(f"Server '{name}' already exists. Overwrite?", default=False): + _info("Cancelled.") + return - # Build initial config if url: server_config["url"] = url else: @@ -504,56 +477,11 @@ def cmd_mcp_add(args): if not _validate_or_warn(name, server_config): return - - # ── Authentication ──────────────────────────────────────────────── - - if url and auth_type == "oauth": - print() - _info(f"Starting OAuth flow for '{name}'...") - oauth_ok = False - try: - from tools.mcp_oauth_manager import get_manager - oauth_auth = get_manager().get_or_build_provider( - name, url, server_config.get("oauth") - ) - if oauth_auth: - server_config["auth"] = "oauth" - _success("OAuth configured (tokens will be acquired on first connection)") - oauth_ok=True - else: - _warning("OAuth setup failed — MCP SDK auth module not available") - except Exception as exc: - _warning(f"OAuth error: {exc}") - - if not oauth_ok: - _info("This server may not support OAuth.") - # Don't store auth: oauth — server doesn't support it - if not _confirm("Continue without authentication?", default=True): - _info("Cancelled.") - return - - elif url: - # Prompt for API key / Bearer token for HTTP servers - print() - _info(f"Connecting to {url}") - needs_auth = _confirm("Does this server require authentication?", default=True) - if needs_auth: - if auth_type == "header" or not auth_type: - env_key = _env_key_for_server(name) - if get_env_value(env_key): - _success(f"{env_key}: already configured") - server_config["headers"] = _bearer_auth_headers(name) # env-var interpolation - else: - api_key = _prompt("API key / Bearer token", password=True) - if api_key: - server_config["headers"] = _save_bearer_auth_token(name, api_key) - _success(f"Saved to {display_hermes_home()}/.env as {env_key}") - - # ── Discovery: connect and list tools ───────────────────────────── + if url and not _configure_http_auth(name, url, auth_type, server_config): + return print() print(color(f" Connecting to '{name}'...", Colors.CYAN)) - try: tools = _probe_single_server(name, server_config) except Exception as exc: @@ -567,66 +495,17 @@ def cmd_mcp_add(args): if not tools: _warning("Server connected but reported no tools.") - if _confirm("Save config anyway?", default=True): - if _save_mcp_server(name, server_config): - _success(f"Saved '{name}' to config") + if _confirm("Save config anyway?", default=True) and _save_mcp_server(name, server_config): + _success(f"Saved '{name}' to config") return - # ── Tool selection ──────────────────────────────────────────────── - - print() - _success(f"Connected! Found {len(tools)} tool(s) from '{name}':") - print() - for tool_name, desc in tools: - short = desc[:60] + "..." if len(desc) > 60 else desc - print(f" {color(tool_name, Colors.GREEN):40s} {short}") - print() - - # Ask: enable all, select, or cancel - try: - choice = input( - color(f" Enable all {len(tools)} tools? [Y/n/select]: ", Colors.YELLOW) - ).strip().lower() - except (KeyboardInterrupt, EOFError): - print() - _info("Cancelled.") + tool_count = _choose_tools(name, tools, server_config) + if tool_count is None: return - - if choice in {"n", "no"}: - _info("Cancelled — server not saved.") - return - - if choice in {"s", "select"}: - # Interactive tool selection - from hermes_cli.curses_ui import curses_checklist - - labels = [f"{t[0]} — {t[1]}" for t in tools] - pre_selected = set(range(len(tools))) - - chosen = curses_checklist( - f"Select tools for '{name}'", - labels, - pre_selected, - ) - - if not chosen: - _info("No tools selected — server not saved.") - return - - chosen_names = [tools[i][0] for i in sorted(chosen)] - server_config.setdefault("tools", {})["include"] = chosen_names - tool_count = len(chosen_names) - else: - # Enable all (no filter needed — default behaviour) - tool_count = len(tools) - total = len(tools) - - # ── Save ────────────────────────────────────────────────────────── - server_config["enabled"] = True if _save_mcp_server(name, server_config): print() - _success(f"Saved '{name}' to {display_hermes_home()}/config.yaml ({tool_count}/{total} tools enabled)") + _success(f"Saved '{name}' to {display_hermes_home()}/config.yaml ({tool_count}/{len(tools)} tools enabled)") _info("Start a new session to use these tools.") @@ -637,17 +516,13 @@ def cmd_mcp_remove(args): name = args.name if _lookup_server(name, _get_mcp_servers()) is None: return - if not _confirm(f"Remove server '{name}'?", default=True): _info("Cancelled.") return - _remove_mcp_server(name) _success(f"Removed '{name}' from config") - - # Clean up OAuth tokens if they exist — route through MCPOAuthManager so - # any provider instance cached in the current process (e.g. from an - # earlier `hermes mcp test` in the same session) is evicted too. + # Route OAuth cleanup through MCPOAuthManager so any provider cached in this process (e.g. from + # an earlier `hermes mcp test`) is evicted too. try: from tools.mcp_oauth_manager import get_manager get_manager().remove(name) @@ -661,7 +536,6 @@ def cmd_mcp_remove(args): def cmd_mcp_list(args=None): """List all configured MCP servers.""" servers = _get_mcp_servers() - if not servers: print() _info("No MCP servers configured.") @@ -675,13 +549,10 @@ def cmd_mcp_list(args=None): print() print(color(" MCP Servers:", Colors.CYAN + Colors.BOLD)) print() - - # Table header print(f" {'Name':<16} {'Transport':<30} {'Tools':<12} {'Status':<10}") print(f" {'─' * 16} {'─' * 30} {'─' * 12} {'─' * 10}") for name, cfg in servers.items(): - # Transport info if "url" in cfg: transport = cfg["url"] elif "command" in cfg: @@ -694,23 +565,14 @@ def cmd_mcp_list(args=None): if len(transport) > 28: transport = transport[:25] + "..." - # Tool count include, exclude = _tool_filters(cfg) - if include: - tools_str = f"{len(include)} selected" - elif exclude: - tools_str = f"-{len(exclude)} excluded" - else: - tools_str = "all" + tools_str = f"{len(include)} selected" if include else f"-{len(exclude)} excluded" if exclude else "all" - # Enabled status enabled = cfg.get("enabled", True) if isinstance(enabled, str): enabled = enabled.lower() in {"true", "1", "yes"} status = color("✓ enabled", Colors.GREEN) if enabled else color("✗ disabled", Colors.DIM) - print(f" {name:<16} {transport:<30} {tools_str:<12} {status}") - print() @@ -724,50 +586,35 @@ def cmd_mcp_test(args): return print() print(color(f" Testing '{name}'...", Colors.CYAN)) - - # Show transport info if "url" in cfg: _info(f"Transport: HTTP → {cfg['url']}") else: - cmd = cfg.get("command", "?") - _info(f"Transport: stdio → {cmd}") + _info(f"Transport: stdio → {cfg.get('command', '?')}") - # Show auth info (masked) - auth_type = cfg.get("auth", "") headers = cfg.get("headers", {}) - if auth_type == "oauth": + if cfg.get("auth", "") == "oauth": _info("Auth: OAuth 2.1 PKCE") elif headers: for k, v in headers.items(): if isinstance(v, str) and ("key" in k.lower() or "auth" in k.lower()): # Mask the value (accepts ${VAR} and Cursor-style ${env:VAR}) resolved = _ENV_VAR_PATTERN.sub(lambda m: os.getenv(_env_ref_name(m.group(1)), ""), v) - if len(resolved) > 8: - masked = resolved[:4] + "***" + resolved[-4:] - else: - masked = "***" + masked = resolved[:4] + "***" + resolved[-4:] if len(resolved) > 8 else "***" print(f" {k}: {masked}") else: _info("Auth: none") - # Attempt connection start = time.monotonic() try: tools = _probe_single_server(name, cfg) - elapsed_ms = (time.monotonic() - start) * 1000 except Exception as exc: - elapsed_ms = (time.monotonic() - start) * 1000 - _error(f"Connection failed ({elapsed_ms:.0f}ms): {exc}") + _error(f"Connection failed ({(time.monotonic() - start) * 1000:.0f}ms): {exc}") return - - _success(f"Connected ({elapsed_ms:.0f}ms)") + _success(f"Connected ({(time.monotonic() - start) * 1000:.0f}ms)") _success(f"Tools discovered: {len(tools)}") - if tools: print() - for tool_name, desc in tools: - short = desc[:55] + "..." if len(desc) > 55 else desc - print(f" {color(tool_name, Colors.GREEN):36s} {short}") + _print_tools(tools, 36, 55) print() @@ -777,8 +624,7 @@ def _reauth_oauth_server(name: str, server_config: dict) -> bool: """Force a fresh OAuth flow for one server. Returns True on success. Wipes cached OAuth state (disk + in-process MCPOAuthManager cache), re-probes to trigger the - browser flow, and verifies a token actually landed before reporting success. Shared by ``hermes - mcp login`` and ``hermes mcp reauth`` so both behave identically for a single server. + browser flow, and verifies a token actually landed. Shared by ``login`` and ``reauth``. """ url = server_config.get("url") if not url: @@ -789,8 +635,6 @@ def _reauth_oauth_server(name: str, server_config: dict) -> bool: _info("Use `hermes mcp remove` + `hermes mcp add` to reconfigure auth.") return False - # Wipe both disk and in-memory cache so the next probe forces a fresh - # OAuth flow. try: from tools.mcp_oauth_manager import get_manager get_manager().remove(name) @@ -800,43 +644,25 @@ def _reauth_oauth_server(name: str, server_config: dict) -> bool: print() _info(f"Starting OAuth flow for '{name}'...") - # Probe triggers the OAuth flow (browser redirect + callback capture). - # Honor the server's configured connect_timeout so a human has enough - # time to complete the browser sign-in; the 30s default is too tight for - # an interactive OAuth round-trip. Floor at 315s — the OAuth callback - # window (300s in mcp_oauth) plus headroom — matching the GUI re-auth - # path in web_server.py so CLI and dashboard behave identically. - # - # force_interactive_oauth: `hermes mcp login` is *explicitly* user- - # initiated even when stdin isn't a TTY (Hermes desktop / agent- - # spawned terminals). Without this, OAuth refuses before opening a - # browser because _is_interactive() only checks sys.stdin.isatty(). + # The probe triggers the OAuth flow (browser redirect + callback capture). Honor the configured + # connect_timeout, floored at 315s (the 300s OAuth callback window + headroom) — matching the GUI + # re-auth path in web_server.py. force_interactive_oauth: `hermes mcp login` is explicitly + # user-initiated even when stdin isn't a TTY (desktop / agent-spawned terminals), where + # _is_interactive() alone would refuse to open a browser. try: from tools.mcp_oauth import force_interactive_oauth - _login_connect_timeout = server_config.get("connect_timeout") try: - _login_connect_timeout = float(_login_connect_timeout) + _login_connect_timeout = float(server_config.get("connect_timeout")) except (TypeError, ValueError): _login_connect_timeout = 0.0 - _login_connect_timeout = max(_login_connect_timeout, 315.0) with force_interactive_oauth(): - tools = _probe_single_server( - name, server_config, connect_timeout=_login_connect_timeout - ) - # A clean probe is NOT proof of authentication. Some MCP servers - # (notably Google's official Drive server) serve initialize + - # tools/list WITHOUT auth, so the probe lists tools even when the - # OAuth flow never completed — e.g. dynamic client registration - # 400'd because the provider doesn't support RFC 7591. Reporting - # "Authenticated — N tools" in that case is a false success: every - # real tool call later hangs until timeout because there's no token. - # Verify a token actually landed on disk before claiming success. + tools = _probe_single_server(name, server_config, connect_timeout=max(_login_connect_timeout, 315.0)) + # A clean probe is NOT proof of authentication: some servers (e.g. Google Drive) serve + # initialize + tools/list without auth, so the flow may have failed (e.g. DCR 400 for + # providers without RFC 7591) while the probe still lists tools. Verify a token landed. if not _oauth_tokens_present(name): - _warning( - "Server responded, but no OAuth token was obtained — " - "authentication did not complete." - ) + _warning("Server responded, but no OAuth token was obtained — authentication did not complete.") print() _info( "Some providers (e.g. Google Drive, Atlassian) do not support " @@ -844,13 +670,11 @@ def _reauth_oauth_server(name: str, server_config: dict) -> bool: "OAuth client yourself and add its credentials to config.yaml:" ) print() - print(color(" mcp_servers:", Colors.DIM)) - print(color(f" {name}:", Colors.DIM)) - print(color(f" url: {url}", Colors.DIM)) - print(color(" auth: oauth", Colors.DIM)) - print(color(" oauth:", Colors.DIM)) - print(color(" client_id: \"\"", Colors.DIM)) - print(color(" client_secret: \"\"", Colors.DIM)) + for line in ( + "mcp_servers:", f" {name}:", f" url: {url}", " auth: oauth", " oauth:", + ' client_id: ""', ' client_secret: ""', + ): + print(color(f" {line}", Colors.DIM)) print() _info("Then re-run `hermes mcp login " + name + "`.") return False @@ -862,10 +686,7 @@ def _reauth_oauth_server(name: str, server_config: dict) -> bool: except Exception as exc: try: from tools.mcp_oauth import humanize_oauth_registration_error - - humanized = humanize_oauth_registration_error( - name, exc, server_url=url - ) + humanized = humanize_oauth_registration_error(name, exc, server_url=url) except Exception: humanized = None _error(f"Authentication failed: {humanized or exc}") @@ -873,32 +694,21 @@ def _reauth_oauth_server(name: str, server_config: dict) -> bool: def cmd_mcp_login(args): - """Force re-authentication for an OAuth-based MCP server. - - Deletes cached tokens (both on disk and in the running process's MCPOAuthManager cache) and - triggers a fresh OAuth flow via the existing probe path. - """ - name = args.name - cfg = _lookup_server(name, _get_mcp_servers()) + """Force re-authentication for an OAuth-based MCP server (wipes cached tokens, re-runs the flow).""" + cfg = _lookup_server(args.name, _get_mcp_servers()) if cfg is not None: - _reauth_oauth_server(name, cfg) + _reauth_oauth_server(args.name, cfg) def cmd_mcp_reauth(args): """Re-authenticate one OAuth MCP server, or all of them sequentially. - Serial-by-design: a human can only complete one browser OAuth flow at a time, so re-authing all - servers concurrently would open N tabs at once and N-1 would time out. + Serial-by-design: a human can only complete one browser OAuth flow at a time. """ servers = _get_mcp_servers() - do_all = getattr(args, "all", False) name = getattr(args, "name", None) - - if do_all: - oauth_servers = [ - (n, c) for n, c in servers.items() - if c.get("auth") == "oauth" and c.get("url") - ] + if getattr(args, "all", False): + oauth_servers = [(n, c) for n, c in servers.items() if c.get("auth") == "oauth" and c.get("url")] if not oauth_servers: _info("No OAuth-based MCP servers found in config.") return @@ -913,7 +723,6 @@ def cmd_mcp_reauth(args): print() _success(f"Re-authenticated {succeeded}/{len(oauth_servers)} server(s)") return - if not name: _error("Specify a server name, or use --all to re-auth every OAuth server.") _info("Usage: hermes mcp reauth | hermes mcp reauth --all") @@ -925,6 +734,36 @@ def cmd_mcp_reauth(args): # ─── hermes mcp configure ──────────────────────────────────────────────────── +def _rebuild_exclude_list(name: str, exclude: list, tool_names: List[str], chosen: set, matches_name_filter) -> List[str]: + """New ``tools.exclude`` for an exclude-mode entry after a checklist edit. + + Stays in exclude mode rather than demoting the user's dynamic filter to a frozen include list: + newly-unchecked tools are appended as literal excludes, re-checked tools drop their literal + entries, and glob patterns are preserved (they keep excluding future vendor tools by design). + """ + old_exclude = [str(p) for p in exclude] + glob_entries = [p for p in old_exclude if "*" in p or "?" in p or "[" in p] + literal_entries = {p for p in old_exclude if p not in glob_entries} + unchecked = {tool_names[i] for i in range(len(tool_names)) if i not in chosen} + checked = {tool_names[i] for i in chosen} + new_literals = (literal_entries - checked) | { + tn for tn in unchecked if not matches_name_filter(tn, set(old_exclude)) + } + # A re-checked tool still matched by a kept glob can't be enabled without dropping the glob — + # surface that instead of silently ignoring the click or silently freezing the config. + glob_shadowed = sorted(tn for tn in checked if glob_entries and matches_name_filter(tn, set(glob_entries))) + if glob_shadowed: + _warning( + f"{len(glob_shadowed)} re-enabled tool(s) still match glob " + f"exclude pattern(s) {glob_entries} and stay excluded: " + f"{', '.join(glob_shadowed[:5])}" + f"{' ...' if len(glob_shadowed) > 5 else ''}. Remove the " + f"pattern from mcp_servers.{name}.tools.exclude in " + "config.yaml to enable them." + ) + return glob_entries + sorted(new_literals) + + def cmd_mcp_configure(args): """Reconfigure which tools are enabled for an existing MCP server.""" import sys as _sys @@ -936,27 +775,22 @@ def cmd_mcp_configure(args): if cfg is None: return - # Discover all available tools print() print(color(f" Connecting to '{name}' to discover tools...", Colors.CYAN)) - try: all_tools = _probe_single_server(name, cfg) except Exception as exc: _error(f"Failed to connect: {exc}") return - if not all_tools: _warning("Server reports no tools.") return - # Determine which are currently enabled include, exclude = _tool_filters(cfg) - tool_names = [t[0] for t in all_tools] + total = len(all_tools) - # Same matching semantics as runtime registration (tools/mcp_tool.py): - # exact names or fnmatch globs. + # Same matching semantics as runtime registration (tools/mcp_tool.py): exact names or globs. try: from tools.mcp_tool import matches_name_filter except ImportError: # pragma: no cover — defensive fallback @@ -965,86 +799,31 @@ def cmd_mcp_configure(args): if include: include_set = {str(p) for p in include} - pre_selected = { - i for i, tn in enumerate(tool_names) - if matches_name_filter(tn, include_set) - } + pre_selected = {i for i, tn in enumerate(tool_names) if matches_name_filter(tn, include_set)} elif exclude: exclude_set = {str(p) for p in exclude} - pre_selected = { - i for i, tn in enumerate(tool_names) - if not matches_name_filter(tn, exclude_set) - } + pre_selected = {i for i, tn in enumerate(tool_names) if not matches_name_filter(tn, exclude_set)} else: - pre_selected = set(range(len(all_tools))) + pre_selected = set(range(total)) - currently = len(pre_selected) - total = len(all_tools) - _info(f"Currently {currently}/{total} tools enabled for '{name}'.") + _info(f"Currently {len(pre_selected)}/{total} tools enabled for '{name}'.") print() - # Interactive checklist from hermes_cli.curses_ui import curses_checklist - labels = [f"{t[0]} — {t[1]}" for t in all_tools] - - chosen = curses_checklist( - f"Select tools for '{name}'", - labels, - pre_selected, - ) - + chosen = curses_checklist(f"Select tools for '{name}'", [f"{t[0]} — {t[1]}" for t in all_tools], pre_selected) if chosen == pre_selected: _info("No changes made.") return - # Update config config = load_config() server_entry = cfg_get(config, "mcp_servers", name, default={}) - exclude_mode = bool(exclude) and not include if len(chosen) == total and not exclude_mode: - # All selected → remove include/exclude (register all) - server_entry.pop("tools", None) + server_entry.pop("tools", None) # all selected → register all elif exclude_mode: - # Exclude-mode entry (catalog default_excluded or hand-written - # tools.exclude): stay in exclude mode instead of demoting the - # user's dynamic filter to a frozen include list. Newly-unchecked - # tools are appended as literal excludes; re-checked tools have - # their literal entries dropped. Glob patterns are preserved — - # they keep excluding future vendor tools by design. - old_exclude = [str(p) for p in (exclude or [])] - glob_entries = [p for p in old_exclude - if "*" in p or "?" in p or "[" in p] - literal_entries = {p for p in old_exclude if p not in glob_entries} - unchecked = {tool_names[i] for i in range(total) if i not in chosen} - checked = {tool_names[i] for i in chosen} - - # Literal excludes: drop re-checked, add newly-unchecked. - new_literals = (literal_entries - checked) | { - tn for tn in unchecked - if not matches_name_filter(tn, set(old_exclude)) - } - new_exclude = glob_entries + sorted(new_literals) - - # A re-checked tool still matched by a kept glob can't be enabled - # without dropping the glob — surface that instead of silently - # ignoring the click or silently freezing the config. - glob_shadowed = sorted( - tn for tn in checked - if glob_entries and matches_name_filter(tn, set(glob_entries)) - ) - if glob_shadowed: - _warning( - f"{len(glob_shadowed)} re-enabled tool(s) still match glob " - f"exclude pattern(s) {glob_entries} and stay excluded: " - f"{', '.join(glob_shadowed[:5])}" - f"{' ...' if len(glob_shadowed) > 5 else ''}. Remove the " - f"pattern from mcp_servers.{name}.tools.exclude in " - "config.yaml to enable them." - ) - + new_exclude = _rebuild_exclude_list(name, exclude, tool_names, chosen, matches_name_filter) if not new_exclude: server_entry.pop("tools", None) else: @@ -1052,32 +831,43 @@ def cmd_mcp_configure(args): server_entry["tools"]["exclude"] = new_exclude server_entry["tools"].pop("include", None) else: - chosen_names = [tool_names[i] for i in sorted(chosen)] server_entry.setdefault("tools", {}) - server_entry["tools"]["include"] = chosen_names + server_entry["tools"]["include"] = [tool_names[i] for i in sorted(chosen)] server_entry["tools"].pop("exclude", None) config.setdefault("mcp_servers", {})[name] = server_entry save_config(config) - - new_count = len(chosen) - _success(f"Updated config: {new_count}/{total} tools enabled") + _success(f"Updated config: {len(chosen)}/{total} tools enabled") _info("Start a new session for changes to take effect.") # ─── Dispatcher ─────────────────────────────────────────────────────────────── +_MCP_USAGE = ( + "hermes mcp Open the catalog picker (default)", + "hermes mcp catalog List Nous-approved MCPs", + "hermes mcp install Install a catalog MCP", + "hermes mcp serve Run as MCP server", + "hermes mcp add --url Add a custom MCP server", + "hermes mcp add --command Add a stdio server", + "hermes mcp add --preset Add from a known preset", + "hermes mcp remove Remove a server", + "hermes mcp list List configured servers", + "hermes mcp test Test connection", + "hermes mcp configure Toggle tools", + "hermes mcp login Re-authenticate OAuth", + "hermes mcp reauth | --all Re-auth one or all OAuth servers", +) + + def mcp_command(args): """Main dispatcher for ``hermes mcp`` subcommands.""" action = getattr(args, "mcp_action", None) - if action == "serve": from mcp_serve import run_mcp_server run_mcp_server(verbose=getattr(args, "verbose", False)) return - - # Catalog subcommands live in mcp_picker / mcp_catalog. Import lazily so - # the original `mcp_config` module stays import-cheap. + # Catalog subcommands live in mcp_picker / mcp_catalog; import lazily to keep this module cheap. if action == "picker": from hermes_cli.mcp_picker import run_picker run_picker() @@ -1093,40 +883,18 @@ def mcp_command(args): if rc: _sys.exit(rc) return - - handlers = { - "add": cmd_mcp_add, - "remove": cmd_mcp_remove, - "rm": cmd_mcp_remove, - "list": cmd_mcp_list, - "ls": cmd_mcp_list, - "test": cmd_mcp_test, - "configure": cmd_mcp_configure, - "config": cmd_mcp_configure, - "login": cmd_mcp_login, - "reauth": cmd_mcp_reauth, - } - - handler = handlers.get(action) + handler = { + "add": cmd_mcp_add, "remove": cmd_mcp_remove, "rm": cmd_mcp_remove, "list": cmd_mcp_list, + "ls": cmd_mcp_list, "test": cmd_mcp_test, "configure": cmd_mcp_configure, + "config": cmd_mcp_configure, "login": cmd_mcp_login, "reauth": cmd_mcp_reauth, + }.get(action) if handler: handler(args) - else: - # No subcommand — drop the user into the catalog picker. This is the - # "try enabling and it flows you into setup" UX matching `hermes plugin`. - from hermes_cli.mcp_picker import run_picker - run_picker() - print(color(" Commands:", Colors.CYAN)) - _info("hermes mcp Open the catalog picker (default)") - _info("hermes mcp catalog List Nous-approved MCPs") - _info("hermes mcp install Install a catalog MCP") - _info("hermes mcp serve Run as MCP server") - _info("hermes mcp add --url Add a custom MCP server") - _info("hermes mcp add --command Add a stdio server") - _info("hermes mcp add --preset Add from a known preset") - _info("hermes mcp remove Remove a server") - _info("hermes mcp list List configured servers") - _info("hermes mcp test Test connection") - _info("hermes mcp configure Toggle tools") - _info("hermes mcp login Re-authenticate OAuth") - _info("hermes mcp reauth | --all Re-auth one or all OAuth servers") - print() + return + # No subcommand — drop the user into the catalog picker (same UX as `hermes plugin`). + from hermes_cli.mcp_picker import run_picker + run_picker() + print(color(" Commands:", Colors.CYAN)) + for line in _MCP_USAGE: + _info(line) + print()