From 3df6a09d62cd082154295d4ee2704cd8446f4f46 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:48:27 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20tools=5Fconfig=5Fpost?= =?UTF-8?q?=5Fsetup/mcp/toolset=5Fvalidation=20=E2=80=94=20shared=20=5Frun?= =?UTF-8?q?=5Ftext,=20flattened=20branches?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/tools_config_mcp.py | 29 ++++----- hermes_cli/tools_config_post_setup.py | 86 ++++++++++----------------- hermes_cli/toolset_validation.py | 45 ++++++-------- 3 files changed, 59 insertions(+), 101 deletions(-) diff --git a/hermes_cli/tools_config_mcp.py b/hermes_cli/tools_config_mcp.py index 12fdcbee20..7606393e51 100644 --- a/hermes_cli/tools_config_mcp.py +++ b/hermes_cli/tools_config_mcp.py @@ -5,14 +5,12 @@ from __future__ import annotations from typing import List, Set from hermes_cli.cli_output import ( - print_error as _print_error, - print_info as _print_info, - print_success as _print_success, + print_error as _print_error, print_info as _print_info, print_success as _print_success, print_warning as _print_warning, ) from hermes_cli.colors import Colors, color from hermes_cli.toolset_scope import ( - _TOOLSET_PLATFORM_RESTRICTIONS, toolset_allowed_for_platform as _toolset_allowed_for_platform + _TOOLSET_PLATFORM_RESTRICTIONS, toolset_allowed_for_platform as _toolset_allowed_for_platform, ) @@ -82,9 +80,8 @@ def _apply_mcp_checklist(server_name: str, tools_cfg: dict, tool_names: List[str def _configure_mcp_tools_interactive(config: dict): """Probe each MCP server for its tools, show a per-server curses checklist, and write the result back as ``tools.exclude`` / ``tools.include`` entries in config.yaml.""" - from hermes_cli.tools_config import save_config - from hermes_cli.curses_ui import curses_checklist + from hermes_cli.tools_config import save_config mcp_servers = config.get("mcp_servers") or {} if not mcp_servers: @@ -134,7 +131,6 @@ def _configure_mcp_tools_interactive(config: dict): for tool_name, description in tools: desc_short = description[:70] + "..." if len(description) > 70 else description labels.append(f"{tool_name} ({desc_short})" if desc_short else tool_name) - match = _mcp_match_filter() tool_names = [t[0] for t in tools] include_set = {str(p) for p in include_list} if include_list else None @@ -275,18 +271,16 @@ def tools_disable_enable_command(args): valid_toolsets = {ts_key for ts_key, _, _ in CONFIGURABLE_TOOLSETS} | _get_plugin_toolset_keys() unknown_toolsets = [t for t in toolset_targets if t not in valid_toolsets] - if unknown_toolsets: - for name in unknown_toolsets: - _print_error(f"Unknown toolset '{name}'") - toolset_targets = [t for t in toolset_targets if t in valid_toolsets] + for name in unknown_toolsets: + _print_error(f"Unknown toolset '{name}'") + toolset_targets = [t for t in toolset_targets if t in valid_toolsets] # Reject platform-scoped toolsets on platforms that don't allow them. restricted_targets = [t for t in toolset_targets if not _toolset_allowed_for_platform(t, platform)] - if restricted_targets: - for name in restricted_targets: - allowed = sorted(_TOOLSET_PLATFORM_RESTRICTIONS.get(name) or set()) - _print_error(f"Toolset '{name}' is not available on platform '{platform}' (only: {', '.join(allowed)})") - toolset_targets = [t for t in toolset_targets if t not in restricted_targets] + for name in restricted_targets: + allowed = sorted(_TOOLSET_PLATFORM_RESTRICTIONS.get(name) or set()) + _print_error(f"Toolset '{name}' is not available on platform '{platform}' (only: {', '.join(allowed)})") + toolset_targets = [t for t in toolset_targets if t not in restricted_targets] if toolset_targets: _apply_toolset_change(config, platform, toolset_targets, action) @@ -301,8 +295,7 @@ def tools_disable_enable_command(args): successful = [ t for t in targets - if t not in unknown_toolsets - and t not in restricted_targets + if t not in unknown_toolsets and t not in restricted_targets and (":" not in t or t.split(":")[0] not in failed_servers) ] if successful: diff --git a/hermes_cli/tools_config_post_setup.py b/hermes_cli/tools_config_post_setup.py index f80d4e38da..15e00d4193 100644 --- a/hermes_cli/tools_config_post_setup.py +++ b/hermes_cli/tools_config_post_setup.py @@ -11,14 +11,12 @@ from pathlib import Path from typing import Set from hermes_cli.cli_output import ( - print_error as _print_error, - print_info as _print_info, - print_success as _print_success, + print_error as _print_error, print_info as _print_info, print_success as _print_success, print_warning as _print_warning, ) from hermes_cli.config import get_env_value from hermes_cli.tools_config_cua import ( - _cua_driver_install_ready, _pip_install, _post_setup_no_window_flags, install_cua_driver + _cua_driver_install_ready, _pip_install, _post_setup_no_window_flags, _run_text, install_cua_driver, ) logger = logging.getLogger("hermes_cli.tools_config") @@ -44,10 +42,8 @@ def _ensure_browser_use_cli(*, verbose_hints: bool = False) -> None: else: for line in str(message).splitlines(): _print_warning(f" {line[:200]}") - if shutil.which("uvx"): - _print_info(" Falling back to zero-install runs via `uvx browser-use`") - else: - _print_info(" Install manually: uv tool install browser-use (https://docs.astral.sh/uv/)") + _print_info(" Falling back to zero-install runs via `uvx browser-use`" if shutil.which("uvx") + else " Install manually: uv tool install browser-use (https://docs.astral.sh/uv/)") if verbose_hints: _print_info(" Local Chrome needs remote debugging: chrome://inspect/#remote-debugging") _print_info(" Cloud browsers: browser-use auth login (or set BROWSER_USE_API_KEY)") @@ -72,27 +68,21 @@ def _install_chromium(install_cmd: list[str]) -> None: """Run the agent-browser Chromium install command and report the outcome.""" _print_info(" Installing Chromium (~170MB one-time download)...") try: - result = subprocess.run( - install_cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", - cwd=str(PROJECT_ROOT), timeout=600, creationflags=_post_setup_no_window_flags(), - ) + result = _run_text(install_cmd, cwd=str(PROJECT_ROOT), timeout=600, creationflags=_post_setup_no_window_flags()) if result.returncode == 0: _print_success(" Chromium installed") # Invalidate the cached "missing" flag so later check_browser_requirements() calls see the install. import tools.browser_tool as _bt _bt._cached_chromium_installed = None - else: - _print_warning(" Chromium install failed:") - tail = (result.stderr or result.stdout or "").strip().splitlines()[-3:] - for line in tail: - _print_info(f" {line[:200]}") - _print_info(" Run manually: npx agent-browser install --with-deps") + return + _print_warning(" Chromium install failed:") + for line in (result.stderr or result.stdout or "").strip().splitlines()[-3:]: + _print_info(f" {line[:200]}") except subprocess.TimeoutExpired: _print_warning(" Chromium install timed out (>10min)") - _print_info(" Run manually: npx agent-browser install --with-deps") except Exception as exc: _print_warning(f" Chromium install failed: {exc}") - _print_info(" Run manually: npx agent-browser install --with-deps") + _print_info(" Run manually: npx agent-browser install --with-deps") def _post_setup_agent_browser(post_setup_key: str) -> None: @@ -106,12 +96,8 @@ def _post_setup_agent_browser(post_setup_key: str) -> None: try: # Lazy import so the tools_config UI doesn't pull in browser_tool at import time. from tools.browser_tool import ( - _chromium_installed, - _running_in_docker, - _find_agent_browser, - _resolve_npx_bin, - _is_npx_agent_browser_sentinel, - AGENT_BROWSER_NPX_SPEC, + _chromium_installed, _running_in_docker, _find_agent_browser, _resolve_npx_bin, + _is_npx_agent_browser_sentinel, AGENT_BROWSER_NPX_SPEC, ) except Exception as exc: # pragma: no cover — defensive _print_warning(f" Could not check Chromium status: {exc}") @@ -164,11 +150,8 @@ def _post_setup_camofox() -> None: elif _npm_bin: _print_info(" Installing Camofox browser server...") # Absolute npm path so the .cmd shim executes on Windows; --workspaces=false avoids resolving apps/desktop. - result = subprocess.run( - [_npm_bin, "install", "--silent", "--workspaces=false"], - capture_output=True, text=True, encoding="utf-8", errors="replace", cwd=str(PROJECT_ROOT), - creationflags=_post_setup_no_window_flags(), - ) + result = _run_text([_npm_bin, "install", "--silent", "--workspaces=false"], timeout=None, + cwd=str(PROJECT_ROOT), creationflags=_post_setup_no_window_flags()) if result.returncode == 0: _print_success(" Camofox installed") else: @@ -230,30 +213,30 @@ _PIP_POST_SETUP_HOOKS: dict = { def _post_setup_pip(spec: dict) -> None: """Run one ``_PIP_POST_SETUP_HOOKS`` entry.""" label = spec["label"] - freshly_installed = False + lines = list(spec["always"]) try: __import__(spec["module"]) - _print_success(f" {label} is already installed") + installed = True except ImportError: + installed = False + if installed: + _print_success(f" {label} is already installed") + else: _print_info(f" {spec['installing']}") try: result = _pip_install(spec["args"], timeout=300) - if result.returncode == 0: - _print_success(f" {label} installed") - freshly_installed = True - else: - _print_warning(f" {label} install failed:") - _print_info(f" {(result.stderr or '').strip()[:300]}") - _print_info(f" Run manually: {spec['manual']}") - return except subprocess.TimeoutExpired: _print_warning(f" {label} install timed out (>5min)") _print_info(f" Run manually: {spec['manual']}") return - if freshly_installed: - for line in spec["on_install"]: - _print_info(f" {line}") - for line in spec["always"]: + if result.returncode != 0: + _print_warning(f" {label} install failed:") + _print_info(f" {(result.stderr or '').strip()[:300]}") + _print_info(f" Run manually: {spec['manual']}") + return + _print_success(f" {label} installed") + lines = list(spec["on_install"]) + lines + for line in lines: _print_info(f" {line}") @@ -400,19 +383,12 @@ def valid_post_setup_keys() -> Set[str]: keys: Set[str] = set() for cat in TOOL_CATEGORIES.values(): - for prov in cat.get("providers", []): - ps = prov.get("post_setup") - if ps: - keys.add(ps) + keys.update(ps for prov in cat.get("providers", []) if (ps := prov.get("post_setup"))) for builder in ( _plugin_web_search_providers, _plugin_image_gen_providers, - _plugin_video_gen_providers, _plugin_browser_providers, - ): + _plugin_video_gen_providers, _plugin_browser_providers): try: - for prov in builder(): - ps = prov.get("post_setup") - if ps: - keys.add(ps) + keys.update(ps for prov in builder() if (ps := prov.get("post_setup"))) except Exception: # pragma: no cover — defensive; plugins optional continue return keys diff --git a/hermes_cli/toolset_validation.py b/hermes_cli/toolset_validation.py index be64be272c..ed3c2ddcd8 100644 --- a/hermes_cli/toolset_validation.py +++ b/hermes_cli/toolset_validation.py @@ -14,17 +14,13 @@ def _platform_default_toolset(platform: object) -> str: def _platform_default_is_valid( - platform: object, - default_toolset: str, - is_valid_toolset: Callable[[str], bool], + platform: object, default_toolset: str, is_valid_toolset: Callable[[str], bool], is_allowed_for_platform: Callable[[str, str], bool], ) -> bool: - if is_valid_toolset(default_toolset) and is_allowed_for_platform( - default_toolset, str(platform) - ): + if is_valid_toolset(default_toolset) and is_allowed_for_platform(default_toolset, str(platform)): return True - # Dynamic plugin platforms are resolved by toolsets.resolve_toolset() even - # though their synthesized hermes- name is not in TOOLSETS. + # Dynamic plugin platforms are resolved by toolsets.resolve_toolset() even though their synthesized + # hermes- name is not in TOOLSETS. try: from gateway.platform_registry import platform_registry @@ -34,8 +30,7 @@ def _platform_default_is_valid( def validate_platform_toolsets( - platform_toolsets: object, - is_valid_toolset: Callable[[str], bool], + platform_toolsets: object, is_valid_toolset: Callable[[str], bool], is_allowed_for_platform: Callable[[str, str], bool] = toolset_allowed_for_platform, ) -> List[str]: """Return human-readable warnings for a ``platform_toolsets`` mapping. @@ -54,17 +49,13 @@ def validate_platform_toolsets( valid_count = 0 for platform, raw in platform_toolsets.items(): default = _platform_default_toolset(platform) - default_valid = _platform_default_is_valid( - platform, default, is_valid_toolset, is_allowed_for_platform - ) + default_valid = _platform_default_is_valid(platform, default, is_valid_toolset, is_allowed_for_platform) platform_valid_count = 0 if not isinstance(raw, list): if default_valid: valid_count += 1 platform_valid_count += 1 - fallback_detail = f"falling back to '{default}'" - else: - fallback_detail = f"falling back to unknown default '{default}'" + fallback_detail = f"falling back to '{default}'" if default_valid else f"falling back to unknown default '{default}'" if raw is None: value_detail = "a null toolset value" elif isinstance(raw, str): @@ -82,18 +73,16 @@ def validate_platform_toolsets( for name in raw: if not isinstance(name, str) or not name: continue - if is_valid_toolset(name): - if is_allowed_for_platform(name, str(platform)): - valid_count += 1 - platform_valid_count += 1 - else: - warnings.append( - f"platform '{platform}' references toolset '{name}' " - "which is not available on this platform" - ) - continue - hint = f" — did you mean '{default}'?" if default_valid else "" - warnings.append(f"platform '{platform}' references unknown toolset '{name}'{hint}") + if not is_valid_toolset(name): + hint = f" — did you mean '{default}'?" if default_valid else "" + warnings.append(f"platform '{platform}' references unknown toolset '{name}'{hint}") + elif is_allowed_for_platform(name, str(platform)): + valid_count += 1 + platform_valid_count += 1 + else: + warnings.append( + f"platform '{platform}' references toolset '{name}' which is not available on this platform" + ) if platform_valid_count == 0: reason = "is configured with an empty toolset list" if not raw else "has no valid toolsets configured"