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

Merge branch 'simp/r3-24-C' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:58:16 -07:00
parent 5ce238fcc6 86b65bbacb
commit 2a96c4c8ae
7 changed files with 740 additions and 1226 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/tools_config.py
+257 -352
View File
File diff suppressed because it is too large Load Diff
hermes_cli/tools_config_cua.py
+206 -334
View File
@@ -11,19 +11,47 @@ from pathlib import Path
from typing import List, Optional
from hermes_cli.cli_output import (
print_info as _print_info, print_success as _print_success, print_warning as _print_warning,
)
print_info as _print_info, print_success as _print_success, print_warning as _print_warning)
logger = logging.getLogger("hermes_cli.tools_config")
# Ceiling for one upstream-installer run: must exceed the installer's own stale-lock recovery window
# (_install-rust.sh force-releases a dead holder's lock only after LOCK_STALE_AFTER_SECONDS=600; a shorter
# timeout kills every run before that fires — a permanent wedge). 660s = 600s + 60s headroom.
_CUA_INSTALLER_TIMEOUT = 660
# Grace for draining the installer's pipes after a timeout kill. The kill is best-effort (_reap_after_timeout)
# so the drain must be bounded: a surviving descendant holding the inherited stdout handle would otherwise
# make the read wait on an EOF that never comes. A successful kill closes the pipe at once, so this is free.
_CUA_INSTALLER_DRAIN_GRACE = 15
# Quiet unattended refreshes from ``hermes update``: bounded even when upstream waits on Read-Host or a
# consent prompt; explicit ``computer-use install --upgrade`` keeps the full ceiling. Safe because the
# lock/network preflights make a legitimate long wait impossible here.
_CUA_BACKGROUND_UPDATE_TIMEOUT = 120
# Upstream installer's stale-lock threshold (LOCK_STALE_AFTER_SECONDS in _install-rust.sh), so the
# pre-clear never yanks a lock a live-but-slow install still holds.
_CUA_LOCK_STALE_AFTER = 600
_CUA_INSTALL_PS1_URL = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.ps1"
_CUA_INSTALL_SH_URL = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.sh"
_CUA_MANUAL_README = "https://github.com/trycua/cua/blob/main/libs/cua-driver/README.md"
_UPGRADE_CMD = "hermes computer-use install --upgrade"
def _run_text(cmd: list, *, timeout, capture_output: bool = True, **kwargs) -> subprocess.CompletedProcess:
"""``subprocess.run`` with the utf-8/replace text decoding every helper here uses."""
return subprocess.run(
cmd, capture_output=capture_output, text=True, encoding="utf-8", errors="replace", timeout=timeout, **kwargs,
)
def _post_setup_no_window_flags(*, streams_to_console: bool = False) -> int:
"""Win32 creationflags that stop post-setup children flashing a console.
CREATE_NO_WINDOW hides console grandchildren (npm, pip, powershell) while keeping stdio inheritable
(unlike DETACHED_PROCESS). Returns 0 on POSIX. ``streams_to_console`` children are only hidden when
our own stdout is not a console, so live installer output is never swallowed.
"""
our own stdout is not a console, so live installer output is never swallowed."""
from hermes_cli._subprocess_compat import windows_hide_flags
flags = windows_hide_flags()
@@ -45,11 +73,7 @@ def _cua_driver_cmd() -> str:
def _cua_version_summary(raw: str, *, limit: int = 120) -> str:
"""First non-empty line of ``--version`` output, bounded (an override may print a multi-line banner)."""
for line in (raw or "").splitlines():
text = line.strip()
if text:
return text[:limit]
return ""
return next((line.strip()[:limit] for line in (raw or "").splitlines() if line.strip()), "")
def _resolved_cua_driver_cmd() -> Optional[str]:
@@ -76,7 +100,8 @@ _CUA_DRIVER_CONTRACT_CACHE: dict = {}
def _cua_driver_contract_status(binary: Optional[str] = None) -> dict:
"""Inspect whether an installed driver supports Hermes' runtime contract."""
"""Inspect whether an installed driver supports Hermes' runtime contract (30s cache keyed on the binary's
path/mtime/size fingerprint)."""
import time
from tools.computer_use.cua_backend import cua_driver_runtime_contract_status
@@ -91,14 +116,12 @@ def _cua_driver_contract_status(binary: Optional[str] = None) -> dict:
return cua_driver_runtime_contract_status(resolved)
now = time.monotonic()
if (
_CUA_DRIVER_CONTRACT_CACHE.get("fingerprint") == fingerprint
and now - _CUA_DRIVER_CONTRACT_CACHE.get("checked_at", 0.0) < 30.0
):
return dict(_CUA_DRIVER_CONTRACT_CACHE["state"])
cache = _CUA_DRIVER_CONTRACT_CACHE
if cache.get("fingerprint") == fingerprint and now - cache.get("checked_at", 0.0) < 30.0:
return dict(cache["state"])
state = cua_driver_runtime_contract_status(resolved)
_CUA_DRIVER_CONTRACT_CACHE.update(fingerprint=fingerprint, checked_at=now, state=dict(state))
cache.update(fingerprint=fingerprint, checked_at=now, state=dict(state))
return state
@@ -111,13 +134,13 @@ def _cua_driver_install_ready() -> bool:
def _pip_install(args: List[str], *, timeout: int = 300, capture_output: bool = True):
"""Install Python packages from a post-setup hook.
Order: ``uv pip install`` (fast, needs no pip in the venv), then ``python -m pip install``, then
``python -m ensurepip --upgrade`` and retry pip. The last tier exists because the Windows installer
creates the venv via ``uv venv``, which does NOT seed pip, so bare ``-m pip`` failed on fresh installs.
"""
creates the venv via ``uv venv``, which does NOT seed pip, so bare ``-m pip`` failed on fresh
installs."""
venv_root = Path(sys.executable).parent.parent
uv_env = {**os.environ, "VIRTUAL_ENV": str(venv_root)}
install_flags = _post_setup_no_window_flags(streams_to_console=not capture_output)
# Managed uv first: $HERMES_HOME/bin is never on PATH, so a bare which() misses the uv Hermes installed;
# ensure_uv() (not a pure lookup) because installing uv is in scope during setup.
@@ -126,12 +149,8 @@ def _pip_install(args: List[str], *, timeout: int = 300, capture_output: bool =
uv_bin = ensure_uv()
if uv_bin:
try:
result = subprocess.run(
[uv_bin, "pip", "install", *args],
capture_output=capture_output, text=True, encoding="utf-8", errors="replace", timeout=timeout,
env=uv_env,
creationflags=_post_setup_no_window_flags(streams_to_console=not capture_output),
)
result = _run_text([uv_bin, "pip", "install", *args], capture_output=capture_output, timeout=timeout,
env=uv_env, creationflags=install_flags)
if result.returncode == 0:
return result
# Fall through to pip — uv may have failed for a reason pip can handle.
@@ -141,31 +160,21 @@ def _pip_install(args: List[str], *, timeout: int = 300, capture_output: bool =
pip_cmd = [sys.executable, "-m", "pip"]
try:
# Probe for pip; bootstrap via ensurepip if missing (uv venv lacks it).
probe = subprocess.run(
pip_cmd + ["--version"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15,
creationflags=_post_setup_no_window_flags(),
)
probe = _run_text(pip_cmd + ["--version"], timeout=15, creationflags=_post_setup_no_window_flags())
if probe.returncode != 0:
raise FileNotFoundError("pip not in venv")
except (subprocess.TimeoutExpired, FileNotFoundError):
try:
subprocess.run(
[sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=120, check=True,
creationflags=_post_setup_no_window_flags(),
)
_run_text([sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"], timeout=120, check=True,
creationflags=_post_setup_no_window_flags())
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
# Synthesize a result so callers see a clean failure path.
return subprocess.CompletedProcess(
pip_cmd, returncode=1, stdout="", stderr=f"pip not available and ensurepip failed: {e}",
)
return subprocess.run(
pip_cmd + ["install", *args],
capture_output=capture_output, text=True, encoding="utf-8", errors="replace", timeout=timeout,
creationflags=_post_setup_no_window_flags(streams_to_console=not capture_output),
)
return _run_text(pip_cmd + ["install", *args], capture_output=capture_output, timeout=timeout,
creationflags=install_flags)
# No pre-install release/asset probe: cua-driver-rs releases are all prereleases, which GitHub's
@@ -187,77 +196,62 @@ def _cua_install_target_writable() -> bool:
def _cua_driver_version(binary: str) -> Optional[str]:
"""``<binary> --version`` stdout (possibly ""), or None when the probe itself fails."""
try:
return subprocess.run(
[binary, "--version"],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5, env=_cua_driver_env(),
creationflags=_post_setup_no_window_flags(),
).stdout.strip()
return _run_text([binary, "--version"], timeout=5, env=_cua_driver_env(),
creationflags=_post_setup_no_window_flags()).stdout.strip()
except Exception:
return None
def _confirmed_update_check(driver_cmd: str, require_confirmed_update: bool) -> tuple:
"""Ask the installed driver whether a newer release exists; returns ``(proceed, pin_version)``.
``proceed=False`` means stop with success (already latest, or indeterminate under
``require_confirmed_update``). An old driver (no check-update verb) or offline check yields None:
`hermes update` then keeps the installed version — an indeterminate check must never cost a
multi-minute silent reinstall on every update — while explicit `install --upgrade` falls through.
"""
multi-minute silent reinstall on every update — while explicit `install --upgrade` falls through."""
try:
from tools.computer_use.cua_backend import cua_driver_update_check
_state = cua_driver_update_check()
except Exception:
_state = None
if _state is not None and not _state.get("update_available"):
if _state is None:
if require_confirmed_update:
_print_info(
f" Could not confirm a newer {driver_cmd} release "
"(offline, rate-limited, or driver too old to check); "
"keeping the installed version.")
_print_info(f" Force a refresh with: {_UPGRADE_CMD}")
return False, None
return True, None
if not _state.get("update_available"):
_print_success(
f" {driver_cmd} is already on the latest release "
f"({_state.get('current_version') or 'unknown'})."
)
f"({_state.get('current_version') or 'unknown'}).")
return False, None
if _state is None and require_confirmed_update:
_print_info(
f" Could not confirm a newer {driver_cmd} release "
"(offline, rate-limited, or driver too old to check); "
"keeping the installed version."
)
_print_info(" Force a refresh with: hermes computer-use install --upgrade")
return False, None
confirmed_version = None
if _state is not None and _state.get("update_available"):
# Windows routine upgrades run unattended-safe (stdin closed, version pinned, ceiling
# _CUA_BACKGROUND_UPDATE_TIMEOUT, preflights skip in seconds); only contract repairs and fresh
# installs stay interactive-only, where upstream needs a human (autostart elevation, SmartScreen).
# Pin to the release check-update confirmed: `latest_version` comes from the GitHub Releases API so
# its assets exist, unlike the installer's baked version on `main`, which is bumped before assets
# are published and 404s unpinned. Malformed values are ignored → unpinned fallback.
import re as _re
# Windows routine upgrades run unattended-safe (stdin closed, version pinned, ceiling
# _CUA_BACKGROUND_UPDATE_TIMEOUT, preflights skip in seconds); only contract repairs and fresh
# installs stay interactive-only, where upstream needs a human (autostart elevation, SmartScreen).
# Pin to the release check-update confirmed: `latest_version` comes from the GitHub Releases API so
# its assets exist, unlike the installer's baked version on `main`, which is bumped before assets
# are published and 404s unpinned. Malformed values are ignored → unpinned fallback.
import re as _re
_latest = str(_state.get("latest_version") or "").strip().lstrip("vV")
if _re.fullmatch(r"\d+(\.\d+)*", _latest):
confirmed_version = _latest
return True, confirmed_version
_latest = str(_state.get("latest_version") or "").strip().lstrip("vV")
return True, (_latest if _re.fullmatch(r"\d+(\.\d+)*", _latest) else None)
def install_cua_driver(
upgrade: bool = False,
require_confirmed_update: bool = False,
show_installer_progress: bool = True,
upgrade: bool = False, require_confirmed_update: bool = False, show_installer_progress: bool = True,
) -> bool:
"""Install or refresh the cua-driver binary used by Computer Use.
The upstream installer always pulls the latest release tag, so re-running it is the canonical way to
upgrade. ``upgrade=False`` (toolset enable flow) keeps a compatible installation, repairs an
old/incomplete one and installs when missing; ``upgrade=True`` always refreshes.
"""
old/incomplete one and installs when missing; ``upgrade=True`` always refreshes."""
import platform as _plat
system = _plat.system()
if system not in ("Darwin", "Windows", "Linux"):
if upgrade:
# Silent: `hermes update` calls this for every user.
return False
_print_warning(" Computer Use (cua-driver) is unsupported on this platform; skipping.")
if not upgrade: # silent under `hermes update`, which calls this for every user
_print_warning(" Computer Use (cua-driver) is unsupported on this platform; skipping.")
return False
is_windows = system == "Windows"
@@ -284,7 +278,7 @@ def install_cua_driver(
return False
if not shutil.which(fetch_tool):
_print_warning(f" {fetch_tool} not found — install manually:")
_print_info(" https://github.com/trycua/cua/blob/main/libs/cua-driver/README.md")
_print_info(f" {_CUA_MANUAL_README}")
return False
return _run_cua_driver_installer(label="Installing")
@@ -298,10 +292,8 @@ def install_cua_driver(
# Compatible existing install: no download, just the host-specific setup upstream normally owns.
if binary and not upgrade and not repair_existing:
version = _cua_driver_version(binary)
if version is None:
_print_success(f" {driver_cmd} already installed.")
else:
_print_success(f" {driver_cmd} already installed: {version or 'unknown version'}")
suffix = "" if version is None else f": {version or 'unknown version'}"
_print_success(f" {driver_cmd} already installed{suffix}.")
if is_windows and not _repair_cua_driver_autostart_windows(binary, verbose=False):
_print_warning(" cua-driver is compatible, but Windows autostart repair failed.")
return False
@@ -311,29 +303,20 @@ def install_cua_driver(
if repair_existing:
version = contract.get("version") or "unknown version"
reason = contract.get("reason") or "required runtime features are missing"
_print_warning(
f" Found cua-driver {version}, but Hermes cannot use its current runtime contract: {reason}."
)
if os.environ.get("HERMES_CUA_DRIVER_CMD", "").strip():
_print_info(
" Update the binary selected by HERMES_CUA_DRIVER_CMD, or unset "
"the override and run: hermes computer-use install --upgrade"
)
_print_warning(f" Found cua-driver {version}, but Hermes cannot use its current runtime contract: {reason}.")
if override:
_print_info(f" Update the binary selected by HERMES_CUA_DRIVER_CMD, or unset the override and run: {_UPGRADE_CMD}")
return False
if is_windows and require_confirmed_update:
_print_info(
" Automatic Windows updates cannot safely run cua-driver's interactive repair installer."
)
_print_info(
" Repair it from an interactive terminal with: hermes computer-use install --upgrade"
)
_print_info(" Automatic Windows updates cannot safely run cua-driver's interactive repair installer.")
_print_info(f" Repair it from an interactive terminal with: {_UPGRADE_CMD}")
return False
_print_info(" Repairing it with the current upstream installer.")
# upgrade=True path — refresh to the latest upstream release.
if not _cua_install_target_writable():
_print_info(" /Applications is not writable; skipping cua-driver refresh.")
_print_info(" Run `hermes computer-use install --upgrade` from an admin account to update it.")
_print_info(f" Run `{_UPGRADE_CMD}` from an admin account to update it.")
return bool(binary)
if not shutil.which(fetch_tool):
@@ -350,30 +333,23 @@ def install_cua_driver(
# Missing binary (enabled but never installed, or wiped by a failed install): an automatic Windows
# update must never launch install.ps1, which can demand console/UAC consent the hidden updater
# cannot provide.
_print_info(
" cua-driver is not installed; automatic Windows updates "
"cannot safely run its interactive installer."
)
_print_info(" Install it from an interactive terminal with: hermes computer-use install --upgrade")
_print_info(" cua-driver is not installed; automatic Windows updates cannot safely run its interactive installer.")
_print_info(f" Install it from an interactive terminal with: {_UPGRADE_CMD}")
return False
# Best-effort before/after version display.
before = (_cua_driver_version(binary) or "") if binary else ""
ok = _run_cua_driver_installer(
label="Repairing" if repair_existing else "Refreshing",
verbose=False,
pin_version=confirmed_version,
label="Repairing" if repair_existing else "Refreshing", verbose=False, pin_version=confirmed_version,
show_progress=show_installer_progress,
installer_timeout=_CUA_BACKGROUND_UPDATE_TIMEOUT if require_confirmed_update else None,
)
installer_timeout=_CUA_BACKGROUND_UPDATE_TIMEOUT if require_confirmed_update else None)
if ok and repair_existing:
repaired = _cua_driver_contract_status()
if not repaired.get("ready"):
_print_warning(
" cua-driver was reinstalled, but its runtime contract is still "
f"unusable: {repaired.get('reason') or 'unknown error'}."
)
f"unusable: {repaired.get('reason') or 'unknown error'}.")
_print_info(" Run: hermes computer-use doctor")
return False
if ok and before:
@@ -385,26 +361,6 @@ def install_cua_driver(
return ok
# Ceiling for one upstream-installer run: must exceed the installer's own stale-lock recovery window
# (_install-rust.sh force-releases a dead holder's lock only after LOCK_STALE_AFTER_SECONDS=600; a shorter
# timeout kills every run before that fires — a permanent wedge). 660s = 600s + 60s headroom.
_CUA_INSTALLER_TIMEOUT = 660
# Grace for draining the installer's pipes after a timeout kill. The kill is best-effort (_reap_after_timeout)
# so the drain must be bounded: a surviving descendant holding the inherited stdout handle would otherwise
# make the read wait on an EOF that never comes. A successful kill closes the pipe at once, so this is free.
_CUA_INSTALLER_DRAIN_GRACE = 15
# Quiet unattended refreshes from ``hermes update``: bounded even when upstream waits on Read-Host or a
# consent prompt; explicit ``computer-use install --upgrade`` keeps the full ceiling. Safe because the
# lock/network preflights make a legitimate long wait impossible here.
_CUA_BACKGROUND_UPDATE_TIMEOUT = 120
# Upstream installer's stale-lock threshold (LOCK_STALE_AFTER_SECONDS in _install-rust.sh), so the
# pre-clear never yanks a lock a live-but-slow install still holds.
_CUA_LOCK_STALE_AFTER = 600
def _cua_install_home() -> "Path":
"""Package home shared by the upstream POSIX and Windows installers."""
return Path(os.environ.get("CUA_DRIVER_RS_HOME") or str(Path.home() / ".cua-driver"))
@@ -422,11 +378,9 @@ def _cua_windows_install_lock_file() -> "Path":
def _clear_stale_windows_cua_install_lock() -> None:
"""Delete install.ps1's lock file only when no process still holds it.
install.ps1 locks with ``FileShare::None``; mirror it with a zero-share ``CreateFileW`` probe and
``FILE_FLAG_DELETE_ON_CLOSE`` so an unlocked leftover is removed atomically, with no window in which
a new installer could acquire the file between probe and delete.
"""
a new installer could acquire the file between probe and delete."""
lock_file = _cua_windows_install_lock_file()
try:
if not lock_file.is_file():
@@ -436,46 +390,31 @@ def _clear_stale_windows_cua_install_lock() -> None:
from ctypes import wintypes as _wintypes
# Win32 constants used by install.ps1's FileShare::None equivalent.
delete_access = 0x00010000
generic_read = 0x80000000
generic_write = 0x40000000
open_existing = 3
file_attribute_normal = 0x00000080
file_flag_delete_on_close = 0x04000000
delete_access, generic_read, generic_write = 0x00010000, 0x80000000, 0x40000000
open_existing, file_attribute_normal, file_flag_delete_on_close = 3, 0x00000080, 0x04000000
kernel32 = _ctypes.WinDLL("kernel32", use_last_error=True)
create_file = kernel32.CreateFileW
create_file.argtypes = [
_wintypes.LPCWSTR, _wintypes.DWORD, _wintypes.DWORD, _wintypes.LPVOID,
_wintypes.DWORD, _wintypes.DWORD, _wintypes.HANDLE,
]
_wintypes.DWORD, _wintypes.DWORD, _wintypes.HANDLE]
create_file.restype = _wintypes.HANDLE
close_handle = kernel32.CloseHandle
close_handle.argtypes = [_wintypes.HANDLE]
close_handle.restype = _wintypes.BOOL
handle = create_file(
str(lock_file),
generic_read | generic_write | delete_access,
0, # FileShare::None
None,
open_existing,
file_attribute_normal | file_flag_delete_on_close,
None,
)
invalid_handle = _wintypes.HANDLE(-1).value
if handle == invalid_handle:
str(lock_file), generic_read | generic_write | delete_access, 0, # 0 = FileShare::None
None, open_existing, file_attribute_normal | file_flag_delete_on_close, None)
if handle == _wintypes.HANDLE(-1).value:
logger.debug(
"Windows cua install lock at %s is still held or cannot be removed (winerror %s)",
lock_file, _ctypes.get_last_error(),
)
lock_file, _ctypes.get_last_error())
return
if not close_handle(handle):
logger.debug(
"could not close Windows cua install lock probe at %s (winerror %s)",
lock_file, _ctypes.get_last_error(),
)
lock_file, _ctypes.get_last_error())
return
if lock_file.exists():
logger.debug("Windows cua install lock probe succeeded but %s remains", lock_file)
@@ -489,11 +428,9 @@ def _clear_stale_windows_cua_install_lock() -> None:
def _clear_stale_cua_install_lock() -> None:
"""Best-effort: remove a stale installer lock left by a dead holder.
POSIX stamps the holder pid into ``~/.cua-driver/packages/.install.lock.d/info``; Windows holds
``~/.cua-driver/install.lock`` open with ``FileShare::None``. Clear either artifact up front only
when its platform-specific liveness check proves that no install still holds it.
"""
when its platform-specific liveness check proves that no install still holds it."""
if sys.platform == "win32":
_clear_stale_windows_cua_install_lock()
return
@@ -502,9 +439,8 @@ def _clear_stale_cua_install_lock() -> None:
if not lock_dir.is_dir():
return
holder_pid = None
info = lock_dir / "info"
try:
for line in info.read_text(encoding="utf-8", errors="replace").splitlines():
for line in (lock_dir / "info").read_text(encoding="utf-8", errors="replace").splitlines():
if line.startswith("pid="):
holder_pid = int(line.split("=", 1)[1].strip())
break
@@ -538,23 +474,21 @@ def _clear_stale_cua_install_lock() -> None:
def _cua_install_lock_held() -> bool:
"""True when the upstream installer's lock is held by a LIVE process.
Called after ``_clear_stale_cua_install_lock()``: anything provably stale is already gone, so a
surviving lock artifact means a concurrent (or orphaned-but-alive) install owns it.
"""
surviving lock artifact means a concurrent (or orphaned-but-alive) install owns it."""
try:
if sys.platform == "win32":
lock_file = _cua_windows_install_lock_file()
if not lock_file.is_file():
return False
# install.ps1 holds the file with FileShare::None — any open fails with a sharing violation
# while held. Surviving the stale-clear = held; confirm with an open probe.
try:
with open(lock_file, "r+b"):
return False # opened fine → not held (racy leftover)
except OSError: # sharing violation surfaces as PermissionError
return True
return _cua_install_lock_dir().is_dir()
if sys.platform != "win32":
return _cua_install_lock_dir().is_dir()
lock_file = _cua_windows_install_lock_file()
if not lock_file.is_file():
return False
# install.ps1 holds the file with FileShare::None — any open fails with a sharing violation
# while held. Surviving the stale-clear = held; confirm with an open probe.
try:
with open(lock_file, "r+b"):
return False # opened fine → not held (racy leftover)
except OSError: # sharing violation surfaces as PermissionError
return True
except Exception as e:
logger.debug("cua install lock probe failed: %s", e)
return False
@@ -562,11 +496,9 @@ def _cua_install_lock_held() -> bool:
def _cua_release_endpoint_reachable(timeout: float = 5.0) -> bool:
"""Fast probe: can we reach GitHub's release download host at all?
When github.com is down the installer dies slowly inside its own retries and eats the whole unattended
ceiling; a 5s HEAD decides in seconds. Only a connection-level failure counts as unreachable — any
HTTP response (even 4xx/5xx) proves the path works.
"""
When github.com is down the installer dies slowly inside its own retries and eats the whole
unattended ceiling; a 5s HEAD decides in seconds. Only a connection-level failure counts as
unreachable — any HTTP response (even 4xx/5xx) proves the path works."""
import urllib.error
import urllib.request
@@ -593,8 +525,7 @@ def _cua_driver_autostart_registered_windows() -> bool:
try:
result = subprocess.run(
["schtasks.exe", "/Query", "/TN", "cua-driver-serve"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
)
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10)
except Exception:
return False
return result.returncode == 0
@@ -602,14 +533,10 @@ def _cua_driver_autostart_registered_windows() -> bool:
def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> bool:
"""Best-effort repair for Windows installer autostart quoting failures.
Older install.ps1 builds interpolated the binary path into a PowerShell command string, which split at
the first space. If the scheduled task is missing, retry via Start-Process's structured ``-FilePath`` /
``-ArgumentList`` parameters instead.
"""
if sys.platform != "win32":
return True
if _cua_driver_autostart_registered_windows():
Older install.ps1 builds interpolated the binary path into a PowerShell command string, which split
at the first space. If the scheduled task is missing, retry via Start-Process's structured
``-FilePath`` / ``-ArgumentList`` parameters instead."""
if sys.platform != "win32" or _cua_driver_autostart_registered_windows():
return True
binary = shutil.which(driver_cmd)
@@ -622,20 +549,12 @@ def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> b
"$proc = Start-Process -FilePath $exe "
"-ArgumentList @('autostart','enable') "
"-Verb RunAs -Wait -PassThru -ErrorAction Stop; "
"exit $proc.ExitCode"
)
if verbose:
_print_info(" Registering cua-driver auto-start...")
else:
_print_info(" Repairing cua-driver auto-start registration...")
"exit $proc.ExitCode")
_print_info(" Registering cua-driver auto-start..." if verbose else " Repairing cua-driver auto-start registration...")
try:
result = subprocess.run(
[ps, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_cmd],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=300,
env=_cua_driver_env(),
)
result = _run_text([ps, "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_cmd], timeout=300,
env=_cua_driver_env())
except subprocess.TimeoutExpired:
_print_warning(" cua-driver autostart registration timed out.")
return False
@@ -646,9 +565,8 @@ def _repair_cua_driver_autostart_windows(driver_cmd: str, *, verbose: bool) -> b
if result.returncode == 0:
return True
tail = (result.stderr or result.stdout or "").strip().splitlines()[-3:]
_print_warning(" cua-driver autostart registration failed.")
for line in tail:
for line in (result.stderr or result.stdout or "").strip().splitlines()[-3:]:
_print_info(f" {line[:200]}")
_print_info(" From an elevated shell, run: cua-driver autostart enable")
return False
@@ -661,6 +579,11 @@ def _remove_quietly(path: str) -> None:
pass
_MAC_PERMISSION_PATHS = (
" System Settings > Privacy & Security > Accessibility",
" System Settings > Privacy & Security > Screen Recording")
def _print_cua_platform_notes(is_windows: bool, is_linux: bool, *, fresh_install: bool) -> None:
"""Host-specific follow-up notes after an install or a compatible-install check."""
if is_windows:
@@ -668,18 +591,13 @@ def _print_cua_platform_notes(is_windows: bool, is_linux: bool, *, fresh_install
_print_info(" Windows/SmartScreen may prompt the first time it runs.")
elif is_linux:
_print_warning(" Linux support is alpha.")
elif fresh_install:
_print_info(" IMPORTANT — grant macOS permissions now:")
_print_info(" System Settings > Privacy & Security > Accessibility")
_print_info(" System Settings > Privacy & Security > Screen Recording")
_print_info(" Both must allow the terminal / Hermes process.")
else:
_print_info(" Grant macOS permissions if not done yet:")
_print_info(" System Settings > Privacy & Security > Accessibility")
_print_info(" System Settings > Privacy & Security > Screen Recording")
_CUA_INSTALL_PS1_URL = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.ps1"
_print_info(" IMPORTANT — grant macOS permissions now:" if fresh_install
else " Grant macOS permissions if not done yet:")
for line in _MAC_PERMISSION_PATHS:
_print_info(line)
if fresh_install:
_print_info(" Both must allow the terminal / Hermes process.")
def _kill_installer_tree(proc, *, is_windows: bool) -> None:
@@ -688,63 +606,52 @@ def _kill_installer_tree(proc, *, is_windows: bool) -> None:
try:
if not is_windows:
os.killpg(os.getpgid(proc.pid), _signal.SIGKILL) # windows-footgun: ok — POSIX branch only
else:
# PowerShell may leave download/install helpers alive after its direct process is killed. They
# inherit stdout and can keep communicate() and install.lock wedged, so kill the tree leaf-up.
import psutil as _psutil
return
# PowerShell may leave download/install helpers alive after its direct process is killed. They
# inherit stdout and can keep communicate() and install.lock wedged, so kill the tree leaf-up.
import psutil as _psutil
try:
parent = _psutil.Process(proc.pid)
descendants = parent.children(recursive=True)
except _psutil.NoSuchProcess:
return
except _psutil.Error as e:
logger.debug(
"could not enumerate cua-driver installer tree for pid %s: %s", proc.pid, e
)
proc.kill()
return
try:
parent = _psutil.Process(proc.pid)
descendants = parent.children(recursive=True)
except _psutil.NoSuchProcess:
return
except _psutil.Error as e:
logger.debug("could not enumerate cua-driver installer tree for pid %s: %s", proc.pid, e)
proc.kill()
return
for child in reversed(descendants):
try:
child.kill()
except _psutil.NoSuchProcess:
pass
except _psutil.Error as e:
logger.debug("could not kill cua-driver installer child pid %s: %s", child.pid, e)
for target, pid in [(child, child.pid) for child in reversed(descendants)] + [(parent, proc.pid)]:
try:
parent.kill()
target.kill()
except _psutil.NoSuchProcess:
pass
except _psutil.Error as e:
logger.debug("could not kill cua-driver installer parent pid %s: %s", proc.pid, e)
proc.kill()
what = "parent" if target is parent else "child"
logger.debug("could not kill cua-driver installer %s pid %s: %s", what, pid, e)
if target is parent:
proc.kill()
except (OSError, ProcessLookupError):
proc.kill()
def _reap_after_timeout(proc, *, is_windows: bool) -> None:
"""Kill the installer tree, then drain its pipes under a deadline.
An unbounded drain blocks on an EOF that only arrives when someone kills a surviving descendant by
hand, so ``_CUA_INSTALLER_TIMEOUT`` would stop bounding anything.
"""
hand, so ``_CUA_INSTALLER_TIMEOUT`` would stop bounding anything."""
_kill_installer_tree(proc, is_windows=is_windows)
try:
drained_out, _ = proc.communicate(timeout=_CUA_INSTALLER_DRAIN_GRACE)
# The partial output names WHERE the installer was stuck (lock wait, consent prompt, download).
if drained_out:
logger.warning(
"cua-driver installer timed out; last output before kill:\n%s", drained_out[-2000:],
)
logger.warning("cua-driver installer timed out; last output before kill:\n%s", drained_out[-2000:])
except subprocess.TimeoutExpired:
# Deliberately not closing proc.stdout: communicate()'s reader threads are still blocked on that
# handle and closing it underneath them races; they are daemon threads.
logger.debug(
"cua-driver installer pipes still open %ss after the kill — "
"abandoning the drain, a surviving descendant holds the inherited handle",
_CUA_INSTALLER_DRAIN_GRACE,
)
_CUA_INSTALLER_DRAIN_GRACE)
except (OSError, ValueError) as e:
logger.debug("cua-driver installer drain failed: %s", e)
@@ -755,23 +662,18 @@ def _cua_installer_command(is_windows: bool):
# Mirror the one-liner printed by cua_driver_install_hint().
ps_oneliner = f"irm {_CUA_INSTALL_PS1_URL} | iex"
install_cmd = ["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", ps_oneliner]
manual_hint = f'powershell -NoProfile -ExecutionPolicy Bypass -Command "{ps_oneliner}"'
return install_cmd, manual_hint, None
return install_cmd, f'powershell -NoProfile -ExecutionPolicy Bypass -Command "{ps_oneliner}"', None
# Download-then-exec instead of `bash -c "$(curl …)"`: no shell=True, no command substitution, and the
# script lands in a mkstemp file (unpredictable name, 0600) rather than a fixed /tmp path — avoiding
# both shell injection and a symlink/TOCTOU race. The manual hint stays the upstream one-liner.
import tempfile as _tempfile
install_url = "https://raw.githubusercontent.com/trycua/cua/main/libs/cua-driver/scripts/install.sh"
manual_hint = f'/bin/bash -c "$(curl -fsSL {install_url})"'
manual_hint = f'/bin/bash -c "$(curl -fsSL {_CUA_INSTALL_SH_URL})"'
fd, script_path = _tempfile.mkstemp(prefix="cua-driver-install-", suffix=".sh")
os.close(fd)
try:
dl = subprocess.run(
["curl", "-fsSL", "-o", script_path, install_url],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=120,
)
dl = _run_text(["curl", "-fsSL", "-o", script_path, _CUA_INSTALL_SH_URL], timeout=120)
failure = None if dl.returncode == 0 else (dl.stderr or "").strip()[:200]
except (subprocess.TimeoutExpired, OSError) as e:
failure = str(e)
@@ -784,24 +686,17 @@ def _cua_installer_command(is_windows: bool):
def _unattended_installer_preflight(install_cmd: list, is_windows: bool):
"""Fail FAST on the two conditions that otherwise consume the whole unattended ceiling.
1. Install lock held by a live process — upstream would poll it for up to LOCK_STALE_AFTER_SECONDS=600
before probing the holder (the 11-minute silent hang class).
2. Release host unreachable — the installer would die slowly inside its own retries; a 5s HEAD answers.
Returns the (possibly rewritten) install command, or None to skip this refresh. Explicit
`computer-use install --upgrade` runs never come here and keep upstream's full lock-recovery.
"""
1. Install lock held by a live process — upstream would poll it for up to
LOCK_STALE_AFTER_SECONDS=600 before probing the holder (the 11-minute silent hang class).
2. Release host unreachable — the installer would die slowly inside its own retries; a 5s HEAD
answers. Returns the (possibly rewritten) install command, or None to skip this refresh. Explicit
`computer-use install --upgrade` runs never come here and keep upstream's full lock-recovery."""
if _cua_install_lock_held():
_print_info(
" Another cua-driver install is in progress (upstream "
"install lock is held) — skipping this refresh."
)
_print_info(" If no install is really running, retry with: hermes computer-use install --upgrade")
_print_info(" Another cua-driver install is in progress (upstream install lock is held) — skipping this refresh.")
_print_info(f" If no install is really running, retry with: {_UPGRADE_CMD}")
return None
if not _cua_release_endpoint_reachable():
_print_info(
" github.com is unreachable — skipping cua-driver refresh (will retry on the next update)."
)
_print_info(" github.com is unreachable — skipping cua-driver refresh (will retry on the next update).")
return None
if is_windows:
# -NoAutoStart skips Register-CuaDriverAutostart — the ONLY branch of install.ps1 that self-elevates
@@ -815,18 +710,12 @@ def _unattended_installer_preflight(install_cmd: list, is_windows: bool):
def _run_cua_driver_installer(
label: str = "Installing",
verbose: bool = True,
pin_version: Optional[str] = None,
show_progress: bool = True,
installer_timeout: Optional[float] = None,
) -> bool:
label: str = "Installing", verbose: bool = True, pin_version: Optional[str] = None, show_progress: bool = True,
installer_timeout: Optional[float] = None) -> bool:
"""Run the upstream cua-driver installer for this platform.
The scripts are idempotent: they always download the latest release, so re-running on an
already-installed system performs an upgrade. ``installer_timeout`` lets quiet callers use a shorter
ceiling without weakening the explicit install path's stale-lock recovery window.
"""
ceiling without weakening the explicit install path's stale-lock recovery window."""
import platform as _plat
system = _plat.system()
@@ -839,10 +728,8 @@ def _run_cua_driver_installer(
install_cmd, manual_hint, script_path = prepared
if show_progress:
if verbose:
_print_info(f" {label} cua-driver (background computer-use)...")
else:
_print_info(f"→ {label} cua-driver (Computer Use)...")
_print_info(f" {label} cua-driver (background computer-use)..." if verbose
else f"→ {label} cua-driver (Computer Use)...")
driver_cmd = _cua_driver_cmd()
timeout = _CUA_INSTALLER_TIMEOUT if installer_timeout is None else installer_timeout
@@ -864,53 +751,41 @@ def _run_cua_driver_installer(
# POSIX: own process group so a timeout kill takes out the whole `curl | bash` pipeline (and the exec'd
# _install-rust.sh), not just the outer shell — surviving grandchildren would keep holding the install
# lock and wedge every later run.
popen_kwargs = {}
popen_kwargs: dict = {"shell": False, "env": installer_env}
if not is_windows:
popen_kwargs["start_new_session"] = True
# Non-verbose (`hermes update` refresh): capture the installer's chatty "Next steps" wall and log it
# so a failure stays debuggable. Verbose interactive installs stream live.
if verbose:
popen_kwargs["creationflags"] = _post_setup_no_window_flags(streams_to_console=True)
else:
popen_kwargs.update(
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, encoding="utf-8", errors="replace", creationflags=_post_setup_no_window_flags(),
)
try:
# Non-verbose (`hermes update` refresh): capture the installer's chatty "Next steps" wall and log it
# so a failure stays debuggable. Verbose interactive installs stream live.
if verbose:
proc = subprocess.Popen(
install_cmd, shell=False, env=installer_env,
creationflags=_post_setup_no_window_flags(streams_to_console=True),
**popen_kwargs
)
try:
proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
_reap_after_timeout(proc, is_windows=is_windows)
raise
result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=None, stderr=None)
else:
proc = subprocess.Popen(
install_cmd, shell=False, env=installer_env,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, encoding="utf-8", errors="replace",
creationflags=_post_setup_no_window_flags(),
**popen_kwargs
)
try:
out, _ = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
_reap_after_timeout(proc, is_windows=is_windows)
raise
result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=out, stderr=None)
proc = subprocess.Popen(install_cmd, **popen_kwargs)
try:
communicated = proc.communicate(timeout=timeout)
except subprocess.TimeoutExpired:
_reap_after_timeout(proc, is_windows=is_windows)
raise
out = None if verbose else communicated[0]
result = subprocess.CompletedProcess(install_cmd, proc.returncode, stdout=out, stderr=None)
if not verbose and result.stdout:
# During `hermes update`, sys.stdout is the mirroring _UpdateOutputStream whose `_log` handle is
# ~/.hermes/logs/update.log — write straight to it so the full installer output is kept
# (success AND failure) without echoing it to the terminal.
if result.stdout:
_update_log = getattr(sys.stdout, "_log", None)
if _update_log is not None:
try:
_update_log.write("\n--- cua-driver installer output ---\n" + result.stdout + "\n")
_update_log.flush()
except Exception:
pass
if result.returncode != 0:
logger.debug("cua-driver installer output:\n%s", result.stdout)
_update_log = getattr(sys.stdout, "_log", None)
if _update_log is not None:
try:
_update_log.write("\n--- cua-driver installer output ---\n" + result.stdout + "\n")
_update_log.flush()
except Exception:
pass
if result.returncode != 0:
logger.debug("cua-driver installer output:\n%s", result.stdout)
installed_binary = _resolved_cua_driver_cmd()
if result.returncode == 0 and installed_binary:
if is_windows and not _repair_cua_driver_autostart_windows(installed_binary, verbose=verbose):
@@ -925,10 +800,7 @@ def _run_cua_driver_installer(
except subprocess.TimeoutExpired:
_print_warning(f" cua-driver {label.lower()} timed out after {timeout}s.")
if not is_windows:
_print_info(
" If this repeats, a stale installer lock may be present — "
f"check {_cua_install_lock_dir()}"
)
_print_info(f" If this repeats, a stale installer lock may be present — check {_cua_install_lock_dir()}")
_print_info(f" Re-run manually: {manual_hint}")
return False
except Exception as e:
hermes_cli/tools_config_mcp.py
+17 -31
View File
@@ -5,25 +5,18 @@ 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_warning as _print_warning,
)
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)
def _mcp_match_filter():
"""Runtime name-filter matcher (exact names or fnmatch globs), with a literal fallback.
Must use the SAME semantics as tools/mcp_tool.py registration — a literal ``in`` check renders
glob excludes (e.g. ``*team_member*`` from catalog default_excluded manifests) as if nothing
were excluded.
"""
Must use the SAME semantics as tools/mcp_tool.py registration — a literal ``in`` check renders glob
excludes (e.g. ``*team_member*`` from catalog default_excluded manifests) as if nothing were
excluded."""
try:
from tools.mcp_tool import matches_name_filter
return matches_name_filter
@@ -68,8 +61,7 @@ def _apply_mcp_checklist(server_name: str, tools_cfg: dict, tool_names: List[str
f"tool(s) still match glob exclude pattern(s) "
f"{glob_entries} and stay excluded — edit "
f"mcp_servers.{server_name}.tools.exclude in config.yaml "
"to enable them."
)
"to enable them.")
if new_exclude:
tools_cfg["exclude"] = new_exclude
else:
@@ -83,9 +75,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:
@@ -135,7 +126,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
@@ -276,18 +266,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)
@@ -302,10 +290,8 @@ def tools_disable_enable_command(args):
successful = [
t for t in 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 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:
verb = "Disabled" if action == "disable" else "Enabled"
_print_success(f"{verb}: {', '.join(successful)}")
hermes_cli/tools_config_post_setup.py
+57 -111
View File
@@ -11,17 +11,11 @@ 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_warning as _print_warning,
)
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")
@@ -31,11 +25,9 @@ PROJECT_ROOT = Path(__file__).parent.parent.resolve()
def _ensure_browser_use_cli(*, verbose_hints: bool = False) -> None:
"""Install the Browser Use CLI if it isn't already runnable.
Primary driver engine for EVERY browser backend except Camofox (Firefox-based, no CDP surface).
MANAGED-FIRST: a browser-use on the user's PATH does NOT satisfy this check — only the
Hermes-managed ``$HERMES_HOME/bin`` copy does.
"""
Hermes-managed ``$HERMES_HOME/bin`` copy does."""
_print_info(" Ensuring browser-use CLI (managed install)...")
try:
from tools.browser_use_cli import install_cli
@@ -47,10 +39,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)")
@@ -75,48 +65,34 @@ 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:
"""``agent_browser`` (local Chromium) and ``browserbase`` (cloud rows) hooks.
agent-browser is not a root package.json dependency — it resolves lazily via npx (or a
global/Hermes-managed install), so there is no ``npm install`` step here.
"""
global/Hermes-managed install), so there is no ``npm install`` step here."""
# Every non-Camofox backend drives through the Browser Use CLI — install it here too.
_ensure_browser_use_cli()
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}")
return
@@ -168,11 +144,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:
@@ -189,10 +162,8 @@ def _post_setup_camofox() -> None:
_KITTENTTS_WHEEL_URL = "https://github.com/KittenML/KittenTTS/releases/download/0.8.1/kittentts-0.8.1-py3-none-any.whl"
# Post-setup hooks that only pip-install a Python package. Fields: module (import probe; present → skip
# install), label (status lines), installing (progress line), args (_pip_install args — keep in sync with
# _RESTORABLE_PYTHON_TOOL_DEPENDENCIES), manual ("Run manually:" command on failure/timeout),
# on_install (info lines only after a fresh install), always (info lines whenever the package is present).
# pip-only post-setup hooks: module (import probe), label, installing (progress line), args (keep in sync with
# _RESTORABLE_PYTHON_TOOL_DEPENDENCIES), manual (fallback command), on_install (fresh-install notes), always.
_PIP_POST_SETUP_HOOKS: dict = {
"faster_whisper": {
"module": "faster_whisper", "label": "faster-whisper",
@@ -200,8 +171,7 @@ _PIP_POST_SETUP_HOOKS: dict = {
"args": ["-U", "faster-whisper", "--quiet"], "manual": "uv pip install -U faster-whisper",
"on_install": ("Model sizes: tiny, base (default), small, medium, large-v3",
"Change via stt.local.model in ~/.hermes/config.yaml"),
"always": (),
},
"always": ()},
"kittentts": {
"module": "kittentts", "label": "kittentts",
"installing": "Installing kittentts (~25-80MB model, CPU-only)...",
@@ -209,8 +179,7 @@ _PIP_POST_SETUP_HOOKS: dict = {
"manual": f"uv pip install -U '{_KITTENTTS_WHEEL_URL}' soundfile",
"on_install": ("Voices: Jasper, Bella, Luna, Bruno, Rosie, Hugo, Kiki, Leo",
"Models: KittenML/kitten-tts-nano-0.8-int8 (25MB), micro (41MB), mini (80MB)"),
"always": (),
},
"always": ()},
"piper": {
"module": "piper", "label": "piper-tts",
"installing": "Installing piper-tts (~14MB wheel, voices downloaded on first use)...",
@@ -218,46 +187,43 @@ _PIP_POST_SETUP_HOOKS: dict = {
"on_install": (),
"always": ("Default voice: en_US-lessac-medium (downloaded on first TTS call)",
"Full voice list: https://github.com/OHF-Voice/piper1-gpl/blob/main/docs/VOICES.md",
"Switch voices by setting tts.piper.voice in ~/.hermes/config.yaml"),
},
"Switch voices by setting tts.piper.voice in ~/.hermes/config.yaml")},
"ddgs": {
"module": "ddgs", "label": "ddgs",
"installing": "Installing ddgs (DuckDuckGo search package)...",
"args": ["-U", "ddgs", "--quiet"], "manual": "uv pip install -U ddgs",
"on_install": (),
"always": ("No API key required. DuckDuckGo enforces server-side rate limits.",
"Pair with an extract provider if you also need web_extract."),
},
}
"Pair with an extract provider if you also need web_extract.")}}
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}")
@@ -274,8 +240,7 @@ def _post_setup_spotify() -> None:
_print_info(" Starting Spotify login...")
try:
login_spotify_command(SimpleNamespace(
client_id=None, redirect_uri=None, scope=None, no_browser=False, timeout=None,
))
client_id=None, redirect_uri=None, scope=None, no_browser=False, timeout=None))
_print_success(" Spotify authenticated")
except SystemExit as exc:
# User aborted the wizard or OAuth failed — don't fail the toolset enable.
@@ -315,10 +280,9 @@ def _post_setup_langfuse() -> None:
def _post_setup_xai_grok() -> None:
"""Shared xAI credential bootstrap for any picker row that talks to xAI (TTS, STT, Video Gen,
x_search …). Accepts a SuperGrok-tier OAuth token (preferred — billed to the existing
subscription) or a raw XAI_API_KEY; the rows declare empty env_vars so the auth UX lives here.
"""
"""Shared xAI credential bootstrap for any picker row that talks to xAI (TTS, STT, Video Gen, x_search
…). Accepts a SuperGrok-tier OAuth token (preferred — billed to the existing subscription) or a raw
XAI_API_KEY; the rows declare empty env_vars so the auth UX lives here."""
try:
from hermes_cli.auth import get_xai_oauth_auth_status
oauth_logged_in = bool(get_xai_oauth_auth_status().get("logged_in"))
@@ -347,10 +311,8 @@ def _post_setup_xai_grok() -> None:
choices=[
"Sign in with xAI Grok OAuth (SuperGrok / Premium+) — browser login",
"Paste an xAI API key (console.x.ai)",
"Skip — configure later via `hermes auth add xai-oauth`",
],
default=0,
)
"Skip — configure later via `hermes auth add xai-oauth`"],
default=0)
if idx == 0:
if _run_xai_oauth_login_from_setup():
_print_success(" Logged in — xAI will use these OAuth credentials")
@@ -392,40 +354,30 @@ def _run_post_setup(post_setup_key: str):
def valid_post_setup_keys() -> Set[str]:
"""Return the set of post-setup keys declared by any visible provider (``TOOL_CATEGORIES`` plus
plugin-registered providers). This is the allowlist ``post-setup`` and the dashboard endpoint
validate against, so a caller cannot drive ``_run_post_setup`` with an arbitrary key.
"""
validate against, so a caller cannot drive ``_run_post_setup`` with an arbitrary key."""
from hermes_cli.tools_config import (
TOOL_CATEGORIES,
_plugin_browser_providers,
_plugin_image_gen_providers,
_plugin_video_gen_providers,
_plugin_web_search_providers,
)
_plugin_web_search_providers)
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
def run_post_setup_command(args) -> int:
"""``hermes tools post-setup <key>`` — non-interactive runner the dashboard spawns so the GUI can
drive backend setup without re-implementing install logic. Exit code: 0 ok, 2 unknown key.
"""
"""``hermes tools post-setup <key>`` — non-interactive runner the dashboard spawns so the GUI can drive
backend setup without re-implementing install logic. Exit code: 0 ok, 2 unknown key."""
key = getattr(args, "post_setup_key", None)
if not key:
_print_error("Usage: hermes tools post-setup <key>")
@@ -449,9 +401,7 @@ def run_post_setup_command(args) -> int:
# binary/dependency install (otherwise toggling the toolset on silently skips the hook). Only add an
# entry when the post_setup is the ONLY install side-effect for a no-key provider and the check is
# local, bounded, and import-light.
_POST_SETUP_INSTALLED: dict = {
"cua_driver": lambda: _cua_driver_install_ready(),
}
_POST_SETUP_INSTALLED: dict = {"cua_driver": lambda: _cua_driver_install_ready()}
def _post_setup_already_installed(post_setup_key: str) -> bool:
@@ -483,16 +433,14 @@ _RESTORABLE_PYTHON_TOOL_DEPENDENCIES: dict[str, tuple[str, tuple[str, ...]]] = {
"kittentts": ("kittentts", ("-U", _KITTENTTS_WHEEL_URL, "soundfile")),
"piper": ("piper", ("-U", "piper-tts")),
"ddgs": ("ddgs", ("-U", "ddgs")),
"langfuse": ("langfuse", ("langfuse",)),
}
"langfuse": ("langfuse", ("langfuse",))}
def active_restorable_python_tool_dependencies() -> list[str]:
"""Return ``hermes tools`` Python dependencies present in this runtime."""
return [
name for name, (module_name, _install_args) in _RESTORABLE_PYTHON_TOOL_DEPENDENCIES.items()
if _module_installed(module_name)
]
if _module_installed(module_name)]
def restorable_python_tool_dependency(name: str) -> tuple[str, tuple[str, ...]] | None:
@@ -501,10 +449,9 @@ def restorable_python_tool_dependency(name: str) -> tuple[str, tuple[str, ...]]
def _agent_browser_installed() -> bool:
"""True when everything ``_run_post_setup("agent_browser")`` installs is present: the agent-browser
CLI *and* the Chromium build it drives (or the Lightpanda engine, which needs no Chromium), so
"Run setup" flips to installed only when re-running it would be a no-op.
"""
"""True when everything ``_run_post_setup("agent_browser")`` installs is present: the agent-browser CLI
*and* the Chromium build it drives (or the Lightpanda engine, which needs no Chromium), so "Run
setup" flips to installed only when re-running it would be a no-op."""
from hermes_cli.nous_subscription import _local_browser_runnable
# The hook runs in a spawned process; this probe runs in the long-lived web-server/CLI process whose
@@ -546,5 +493,4 @@ _POST_SETUP_READY: dict = {
"browserbase": lambda: _cloud_agent_browser_installed(),
"camofox": lambda: _camofox_installed(),
"lightpanda": lambda: _lightpanda_installed(),
"cua_driver": lambda: _cua_driver_install_ready(),
}
"cua_driver": lambda: _cua_driver_install_ready()}
hermes_cli/tools_config_providers.py
+177 -349
View File
@@ -4,13 +4,11 @@ from __future__ import annotations
import importlib
import logging
from functools import partial
from typing import Callable, Optional
from hermes_cli.cli_output import (
print_info as _print_info,
print_success as _print_success,
print_warning as _print_warning,
prompt as _prompt,
print_info as _print_info, print_success as _print_success, print_warning as _print_warning, prompt as _prompt,
)
from hermes_cli.colors import Colors, color
from hermes_cli.config import cfg_get, get_env_value, load_config, save_config, save_env_value
@@ -38,19 +36,13 @@ def _plugin_registry(module: str):
def _plugin_provider_rows(
registry_module: str,
marker_keys: tuple[str, ...],
*,
require_name: bool = True,
skip_builtin: bool = False,
flatten_variants: bool = False,
) -> list[dict]:
registry_module: str, marker_keys: tuple[str, ...], *, require_name: bool = True, skip_builtin: bool = False,
flatten_variants: bool = False) -> list[dict]:
"""Picker-row dicts (TOOL_CATEGORIES-shaped) for a plugin registry's providers.
``marker_keys`` are all set to the registry name so downstream config/model-picker code routes through
the registry. ``skip_builtin`` drops names shadowing ``_BUILTIN_NAMES``; ``flatten_variants`` expands a
schema's tier ``variants`` into separate rows sharing one backend name, distinguished by ``web_tier``.
"""
``marker_keys`` are all set to the registry name so downstream config/model-picker code routes
through the registry. ``skip_builtin`` drops names shadowing ``_BUILTIN_NAMES``;
``flatten_variants`` expands a schema's tier ``variants`` into separate rows sharing one backend
name, distinguished by ``web_tier``."""
registry = _plugin_registry(registry_module)
if registry is None:
return []
@@ -89,66 +81,41 @@ def _plugin_provider_rows(
return rows
# Category -> (registry module, marker keys, _plugin_provider_rows kwargs). image/video ``*_plugin_name`` route
# config writes + model pickers through the registry (video has no in-tree backend); all bundled web providers
# are plugins (only firecrawl setup-flow rows stay hardcoded); ``browser_provider`` is the legacy key written to
# ``browser.cloud_provider``; TTS rows render below the built-in rows and write ``tts.provider: <name>``.
# Category -> (registry module, marker keys, _plugin_provider_rows kwargs). ``*_plugin_name`` markers route
# config writes + model pickers through the registry; ``browser_provider`` is the legacy ``browser.cloud_provider``
# key; TTS rows write ``tts.provider: <name>``.
_PLUGIN_PROVIDER_ROW_SPECS = {
"image_gen": ("agent.image_gen_registry", ("image_gen_plugin_name",), {"require_name": False}),
"video_gen": ("agent.video_gen_registry", ("video_gen_plugin_name",), {"require_name": False}),
"web": ("agent.web_search_registry", ("web_backend", "web_search_plugin_name"), {"flatten_variants": True}),
"browser": ("agent.browser_registry", ("browser_provider", "browser_plugin_name"), {}),
"tts": ("agent.tts_registry", ("tts_provider", "tts_plugin_name"), {"skip_builtin": True}),
}
"tts": ("agent.tts_registry", ("tts_provider", "tts_plugin_name"), {"skip_builtin": True})}
def _plugin_rows_for(category: str) -> list[dict]:
"""Picker rows for the plugin-registered providers of one ``_PLUGIN_PROVIDER_ROW_SPECS`` category."""
module, markers, kwargs = _PLUGIN_PROVIDER_ROW_SPECS[category]
return _plugin_provider_rows(module, markers, **kwargs)
def _plugin_image_gen_providers() -> list[dict]:
"""Picker rows for plugin-registered image gen providers."""
return _plugin_rows_for("image_gen")
def _plugin_video_gen_providers() -> list[dict]:
"""Picker rows for plugin-registered video gen providers (the only Video Generation rows)."""
return _plugin_rows_for("video_gen")
def _plugin_web_search_providers() -> list[dict]:
"""Picker rows for plugin-registered web search providers."""
return _plugin_rows_for("web")
def _plugin_browser_providers() -> list[dict]:
"""Picker rows for plugin-registered cloud browser providers."""
return _plugin_rows_for("browser")
def _plugin_tts_providers() -> list[dict]:
"""Picker rows for plugin-registered TTS providers."""
return _plugin_rows_for("tts")
_plugin_image_gen_providers = partial(_plugin_rows_for, "image_gen")
_plugin_video_gen_providers = partial(_plugin_rows_for, "video_gen") # the only Video Generation rows
_plugin_web_search_providers = partial(_plugin_rows_for, "web")
_plugin_browser_providers = partial(_plugin_rows_for, "browser")
_plugin_tts_providers = partial(_plugin_rows_for, "tts")
def web_provider_capabilities(backend: str) -> list:
"""Capabilities (``search`` / ``extract``) a web backend supports, per the registry provider instance.
Lets the Capabilities GUI offer ``web.search_backend`` / ``web.extract_backend`` only where it makes sense
(ddgs and brave-free are search-only). Unknown backend or registry failure -> both.
"""
Lets the Capabilities GUI offer ``web.search_backend`` / ``web.extract_backend`` only where it makes
sense (ddgs and brave-free are search-only). Unknown backend or registry failure -> both."""
try:
from agent.web_search_registry import get_provider
provider = get_provider(backend)
if provider is not None:
caps = []
if provider.supports_search():
caps.append("search")
if provider.supports_extract():
caps.append("extract")
return caps
return [cap for cap, supported in (("search", provider.supports_search()),
("extract", provider.supports_extract())) if supported]
except Exception:
pass
return ["search", "extract"]
@@ -160,22 +127,15 @@ _PLUGIN_ROW_BUILDERS = {
"Video Generation": _plugin_video_gen_providers,
"Web Search & Extract": _plugin_web_search_providers,
"Browser Automation": _plugin_browser_providers,
"Text-to-Speech": _plugin_tts_providers,
}
"Text-to-Speech": _plugin_tts_providers}
def _visible_providers(
cat: dict,
config: dict,
*,
force_fresh: bool = False,
features: Optional[NousSubscriptionFeatures] = None,
cat: dict, config: dict, *, force_fresh: bool = False, features: Optional[NousSubscriptionFeatures] = None,
) -> list[dict]:
"""Provider entries visible for the current auth/config state.
Nous-managed rows (``managed_nous_feature``) are always shown, even logged-out/unentitled, to advertise
the capability.
"""
Nous-managed rows (``managed_nous_feature``) are always shown, even logged-out/unentitled, to
advertise the capability."""
from hermes_cli.tools_config import get_nous_subscription_features
if features is None:
@@ -186,15 +146,12 @@ def _visible_providers(
pool_only = bool(acct and acct.logged_in and acct.paid_service_access is not True and acct.tool_gateway_entitled)
visible = []
for provider in cat.get("providers", []):
managed = provider.get("managed_nous_feature")
# Managed rows stay visible regardless of auth (selecting one drives an inline Portal login); a
# `requires_nous_auth` row without a managed feature hides until logged in.
if provider.get("requires_nous_auth") and not provider.get("managed_nous_feature") and not features.nous_auth_present:
if provider.get("requires_nous_auth") and not managed and not features.nous_auth_present:
continue
if (
pool_only
and provider.get("managed_nous_feature") == "video_gen"
and not (acct and acct.tool_gateway_entitled_for("fal-video"))
):
if pool_only and managed == "video_gen" and not (acct and acct.tool_gateway_entitled_for("fal-video")):
continue
visible.append(provider)
@@ -205,23 +162,12 @@ def _visible_providers(
return visible
def provider_readiness_status(
provider: dict,
config: dict,
*,
features=None,
is_active: Optional[bool] = None,
) -> str:
def provider_readiness_status(provider: dict, config: dict, *, features=None, is_active: Optional[bool] = None) -> str:
"""Honest readiness state for a provider picker row.
``features`` avoids re-fetching portal state per row. ``is_active`` is the completed-setup fallback for
post_setup hooks with no registered installed-check (selecting a row runs its hook).
"""
``features`` avoids re-fetching portal state per row. ``is_active`` is the completed-setup fallback
for post_setup hooks with no registered installed-check (selecting a row runs its hook)."""
from hermes_cli.tools_config import (
_POST_SETUP_READY,
_provider_env_ready,
_xai_credentials_present,
get_nous_subscription_features,
_POST_SETUP_READY, _provider_env_ready, _xai_credentials_present, get_nous_subscription_features,
)
if provider.get("env_vars", []):
@@ -261,12 +207,7 @@ def provider_readiness_status(
return "ready"
def _toolset_needs_configuration_prompt(
ts_key: str,
config: dict,
*,
force_fresh: bool = False,
) -> bool:
def _toolset_needs_configuration_prompt(ts_key: str, config: dict, *, force_fresh: bool = False) -> bool:
"""Return True when enabling this toolset should open provider setup."""
from hermes_cli.tools_config import TOOL_CATEGORIES, _post_setup_already_installed, _toolset_has_keys
@@ -311,19 +252,10 @@ def _any_plugin_provider_available(registry_module: str) -> bool:
return False
def _configure_tool_category(
ts_key: str,
cat: dict,
config: dict,
*,
force_fresh: bool = True,
reconfigure: bool = False,
):
def _configure_tool_category(ts_key: str, cat: dict, config: dict, *, force_fresh: bool = True, reconfigure: bool = False):
"""Provider selection for a tool category, then API-key setup for the chosen row.
``reconfigure`` ("Reconfigure an existing tool"): no setup note / skip row / Nous marker, and the chosen
provider goes through the key-update prompts instead of the new-enable prompts.
"""
``reconfigure`` ("Reconfigure an existing tool"): no setup note / skip row / Nous marker, and the
chosen provider goes through the key-update prompts instead of the new-enable prompts."""
from hermes_cli.tools_config import _prompt_choice, _provider_env_ready, get_nous_subscription_features
icon = cat.get("icon", "")
@@ -387,11 +319,10 @@ def _configure_tool_category(
def _web_tier_matches(provider: dict, config: dict) -> bool:
"""True when a web picker row's tier matches the configured tier (``web.provider_tier.<backend>``).
Tiered rows (Exa/Parallel Free vs Paid) share one ``web_backend`` and differ only in ``web_tier``. No
``web_tier`` on the row → matches; configured tier set → must equal the row's tier; unset → "auto":
paid when the row's key is present, free otherwise (highlight the row the runtime would actually use).
"""
Tiered rows (Exa/Parallel Free vs Paid) share one ``web_backend`` and differ only in ``web_tier``.
No ``web_tier`` on the row → matches; configured tier set → must equal the row's tier; unset →
"auto": paid when the row's key is present, free otherwise (highlight the row the runtime would
actually use)."""
row_tier = provider.get("web_tier")
if not row_tier:
return True
@@ -413,6 +344,18 @@ def _web_tier_matches(provider: dict, config: dict) -> bool:
return row_tier == ("paid" if has_key else "free")
# Managed-row marker -> (config section, key) the pick writes, in check order.
_MANAGED_SELECTION_KEYS: tuple[tuple[str, str, str], ...] = (
("tts_provider", "tts", "provider"), ("stt_provider", "stt", "provider"),
("browser_provider", "browser", "cloud_provider"), ("web_backend", "web", "backend"))
def _has_marker(provider: dict, marker: str) -> bool:
"""``browser_provider`` is a membership test (a local row carries ``browser_provider: ""``); every other
marker is a truthiness test."""
return marker in provider if marker == "browser_provider" else bool(provider.get(marker))
def _managed_provider_active(provider: dict, config: dict, managed_feature: str, force_fresh: bool) -> bool:
"""Active check for a Nous-managed row: the feature must be managed AND the category's selected
provider must be the row's vendor or ``nous``."""
@@ -431,28 +374,18 @@ def _managed_provider_active(provider: dict, config: dict, managed_feature: str,
if (
configured_provider != NOUS_MANAGED_PROVIDER
and gen_cfg.get("use_gateway") is not None
and not is_truthy_value(gen_cfg.get("use_gateway"), default=False)
):
and not is_truthy_value(gen_cfg.get("use_gateway"), default=False)):
return False
return feature.managed_by_nous
if provider.get("tts_provider"):
current = cfg_get(config, "tts", "provider")
return feature.managed_by_nous and current in {provider["tts_provider"], NOUS_MANAGED_PROVIDER}
if provider.get("stt_provider"):
current = cfg_get(config, "stt", "provider")
return feature.managed_by_nous and current in {provider["stt_provider"], NOUS_MANAGED_PROVIDER}
if "browser_provider" in provider:
# Browser Use mode is a driver on top of the provider (attaches to its CDP endpoint), so the
# provider row stays active alongside the Browser Use row.
current = cfg_get(config, "browser", "cloud_provider")
return feature.managed_by_nous and current in {provider["browser_provider"], NOUS_MANAGED_PROVIDER}
if provider.get("web_backend"):
current = cfg_get(config, "web", "backend")
return (
feature.managed_by_nous
and current in {provider["web_backend"], NOUS_MANAGED_PROVIDER}
and _web_tier_matches(provider, config)
)
# Browser Use mode is a driver on top of the provider (attaches to its CDP endpoint), so the browser
# provider row stays active alongside the Browser Use row.
for marker, section, key in _MANAGED_SELECTION_KEYS:
if _has_marker(provider, marker):
current = cfg_get(config, section, key)
selected = current in {provider[marker], NOUS_MANAGED_PROVIDER}
if marker == "web_backend":
selected = selected and _web_tier_matches(provider, config)
return feature.managed_by_nous and selected
return feature.managed_by_nous
@@ -477,15 +410,6 @@ def _browser_use_default_active(config: dict) -> bool:
return False
def _tts_active(provider: dict, config: dict) -> bool:
return cfg_get(config, "tts", "provider") == provider["tts_provider"]
def _stt_active(provider: dict, config: dict) -> bool:
# Default stt.provider is "local" — an unset key means Local Whisper.
return (cfg_get(config, "stt", "provider") or "local") == provider["stt_provider"]
def _browser_provider_active(provider: dict, config: dict) -> bool:
# Browser Use mode composes with the provider (driver over its CDP endpoint) — don't deactivate the row.
if provider["browser_provider"] != cfg_get(config, "browser", "cloud_provider"):
@@ -511,44 +435,28 @@ def _browser_backend_active(provider: dict, config: dict) -> bool:
return _browser_use_default_active(config)
def _web_backend_active(provider: dict, config: dict) -> bool:
return cfg_get(config, "web", "backend") == provider["web_backend"] and _web_tier_matches(provider, config)
def _computer_use_active(provider: dict, config: dict) -> bool:
return cfg_get(config, "computer_use", "backend") == provider["computer_use_backend"]
def _imagegen_backend_active(provider: dict, config: dict) -> bool:
image_cfg = config.get("image_gen", {})
return (
isinstance(image_cfg, dict)
and provider["imagegen_backend"] == "fal"
and image_cfg.get("provider") in {None, "", "fal"}
and not is_truthy_value(image_cfg.get("use_gateway"), default=False)
)
and not is_truthy_value(image_cfg.get("use_gateway"), default=False))
# Non-managed active checks, evaluated in order; the first marker the row carries decides. ``browser_provider``
# is a membership test (a local row carries ``browser_provider: ""``), every other marker is a truthiness test.
# Non-managed active checks, evaluated in order; the first marker the row carries decides (see ``_has_marker``).
# Default stt.provider is "local" — an unset key means Local Whisper.
_ACTIVE_CHECKS: tuple[tuple[str, Callable[[dict, dict], bool]], ...] = (
("tts_provider", _tts_active),
("stt_provider", _stt_active),
("tts_provider", lambda p, c: cfg_get(c, "tts", "provider") == p["tts_provider"]),
("stt_provider", lambda p, c: (cfg_get(c, "stt", "provider") or "local") == p["stt_provider"]),
("browser_provider", _browser_provider_active),
("browser_backend", _browser_backend_active),
("web_backend", _web_backend_active),
("computer_use_backend", _computer_use_active),
("imagegen_backend", _imagegen_backend_active),
)
_MEMBERSHIP_MARKERS = frozenset({"browser_provider"})
("web_backend", lambda p, c: cfg_get(c, "web", "backend") == p["web_backend"] and _web_tier_matches(p, c)),
("computer_use_backend", lambda p, c: cfg_get(c, "computer_use", "backend") == p["computer_use_backend"]),
("imagegen_backend", _imagegen_backend_active))
def _is_provider_active(
provider: dict,
config: dict,
*,
force_fresh: bool = False,
) -> bool:
def _is_provider_active(provider: dict, config: dict, *, force_fresh: bool = False) -> bool:
"""Check if a provider entry matches the currently active config."""
managed_feature = provider.get("managed_nous_feature")
plugin_name = provider.get("image_gen_plugin_name")
@@ -570,18 +478,12 @@ def _is_provider_active(
return _managed_provider_active(provider, config, managed_feature, force_fresh)
for marker, check in _ACTIVE_CHECKS:
present = marker in provider if marker in _MEMBERSHIP_MARKERS else provider.get(marker)
if present:
if _has_marker(provider, marker):
return check(provider, config)
return False
def _detect_active_provider_index(
providers: list,
config: dict,
*,
force_fresh: bool = False,
) -> int:
def _detect_active_provider_index(providers: list, config: dict, *, force_fresh: bool = False) -> int:
"""Return the index of the currently active provider, or 0."""
from hermes_cli.tools_config import _provider_env_ready
@@ -602,8 +504,7 @@ def _fal_model_catalog():
# Per-backend model catalog (config_key = top-level config.yaml section, catalog_fn -> ({model_id: metadata},
# default_model)); a TOOL_CATEGORIES row tagged `imagegen_backend: "<name>"` selects the catalog at picker time.
IMAGEGEN_BACKENDS = {
"fal": {"display": "FAL.ai", "config_key": "image_gen", "catalog_fn": _fal_model_catalog},
}
"fal": {"display": "FAL.ai", "config_key": "image_gen", "catalog_fn": _fal_model_catalog}}
def _plugin_model_catalog(registry_module: str, plugin_name: str):
@@ -627,31 +528,18 @@ def _plugin_model_catalog(registry_module: str, plugin_name: str):
return catalog, default
def _plugin_image_gen_catalog(plugin_name: str):
"""``(catalog, default_model)`` for a plugin-registered image gen provider."""
return _plugin_model_catalog("agent.image_gen_registry", plugin_name)
def _plugin_video_gen_catalog(plugin_name: str):
"""``(catalog, default_model)`` for a plugin-registered video gen provider."""
return _plugin_model_catalog("agent.video_gen_registry", plugin_name)
_plugin_image_gen_catalog = partial(_plugin_model_catalog, "agent.image_gen_registry")
_plugin_video_gen_catalog = partial(_plugin_model_catalog, "agent.video_gen_registry")
def _pick_model_from_catalog(
catalog: dict,
default_model,
cfg_key: str,
display: str,
config: dict,
*,
row_indent: str = "",
catalog: dict, default_model, cfg_key: str, display: str, config: dict, *, row_indent: str = "",
) -> None:
"""Column-aligned model picker shared by the FAL, plugin image gen and video gen flows.
Writes the choice to ``config[cfg_key]["model"]``. The current model is listed first so the cursor lands
on it; a saved model belonging to another provider (shared config key) or a drifted catalog default never
indexes the catalog. Safe when stdin is not a TTY — curses_radiolist keeps the current selection.
"""
Writes the choice to ``config[cfg_key]["model"]``. The current model is listed first so the cursor
lands on it; a saved model belonging to another provider (shared config key) or a drifted catalog
default never indexes the catalog. Safe when stdin is not a TTY — curses_radiolist keeps the current
selection."""
from hermes_cli.tools_config import _cfg_section, _prompt_choice
if not catalog:
@@ -666,8 +554,7 @@ def _pick_model_from_catalog(
widths = {
"model": max(len(m) for m in model_ids),
"speed": max((len(catalog[m].get("speed", "")) for m in model_ids), default=6),
"strengths": max((len(catalog[m].get("strengths", "")) for m in model_ids), default=0),
}
"strengths": max((len(catalog[m].get("strengths", "")) for m in model_ids), default=0)}
print()
header = (f" {'Model':<{widths['model']}} {'Speed':<{widths['speed']}} "
@@ -705,9 +592,8 @@ def _configure_imagegen_model_for_plugin(plugin_name: str, config: dict) -> None
def _configure_videogen_model_for_plugin(plugin_name: str, config: dict) -> None:
"""Prompt for a video gen model from a plugin's catalog."""
"""Prompt for a video gen model from a plugin's catalog (rows keep their historical two-space indent)."""
catalog, default_model = _plugin_video_gen_catalog(plugin_name)
# Video rows historically carry a two-space indent; kept for output parity.
_pick_model_from_catalog(catalog, default_model, "video_gen", plugin_name, config, row_indent=" ")
@@ -715,38 +601,29 @@ def _configure_xai_imagine_storage(section_name: str, config: dict) -> None:
"""Prompt for xAI Imagine stored public URL behavior."""
from hermes_cli.tools_config import _cfg_section, _prompt_choice
section = _cfg_section(config, section_name)
xai_cfg = _cfg_section(section, "xai")
storage_cfg = _cfg_section(xai_cfg, "storage")
storage_cfg = _cfg_section(_cfg_section(_cfg_section(config, section_name), "xai"), "storage")
_print_warning(
" xAI Imagine can store generated media and create reusable public URLs. "
"xAI may bill for stored files and public URL hosting."
)
"xAI may bill for stored files and public URL hosting.")
choices = ["Enable public URLs without automatic expiry (recommended)", "Disable stored public URLs",
"Enable public URLs for 2 days"]
idx = _prompt_choice(" Stored public URLs:", choices, default=0)
if idx == 1:
storage_cfg["enabled"] = False
_print_success(" xAI stored public URLs disabled")
elif idx == 2:
storage_cfg["enabled"] = True
storage_cfg["public_url"] = True
storage_cfg["expires_after"] = 2 * 24 * 60 * 60
_print_success(" xAI stored public URLs enabled for 2 days")
else:
storage_cfg["enabled"] = True
storage_cfg["public_url"] = True
storage_cfg["expires_after"] = None
_print_success(" xAI stored public URLs enabled without automatic expiry")
return
storage_cfg["enabled"] = True
storage_cfg["public_url"] = True
storage_cfg["expires_after"] = 2 * 24 * 60 * 60 if idx == 2 else None
_print_success(" xAI stored public URLs enabled for 2 days" if idx == 2
else " xAI stored public URLs enabled without automatic expiry")
def _select_plugin_gen_provider(section: str, plugin_name: str, config: dict, *, use_gateway: bool) -> None:
def _select_plugin_gen_provider(section: str, plugin_name: str, config: dict, *, use_gateway: bool = False) -> None:
"""Persist a plugin-backed image/video gen provider selection and run its model picker.
``use_gateway=True`` (Nous-managed pick) stores ``<section>.provider: nous``; BYOK picks store the plugin
name. Any legacy ``use_gateway`` key is removed so old read-time shims cannot override the selection.
"""
``use_gateway=True`` (Nous-managed pick) stores ``<section>.provider: nous``; BYOK picks store the
plugin name. Any legacy ``use_gateway`` key is removed so old read-time shims cannot override the
selection."""
from hermes_cli.tools_config import _cfg_section
cfg = _cfg_section(config, section)
@@ -761,13 +638,8 @@ def _select_plugin_gen_provider(section: str, plugin_name: str, config: dict, *,
_configure_xai_imagine_storage(section, config)
def _select_plugin_image_gen_provider(plugin_name: str, config: dict, *, use_gateway: bool = False) -> None:
_select_plugin_gen_provider("image_gen", plugin_name, config, use_gateway=use_gateway)
def _select_plugin_video_gen_provider(plugin_name: str, config: dict, *, use_gateway: bool = False) -> None:
_select_plugin_gen_provider("video_gen", plugin_name, config, use_gateway=use_gateway)
_select_plugin_image_gen_provider = partial(_select_plugin_gen_provider, "image_gen")
_select_plugin_video_gen_provider = partial(_select_plugin_gen_provider, "video_gen")
# Per-provider STT model catalogs for the picker; keys are ``stt.<provider>`` sections, first entry is the
# default. Kept in sync with the dashboard selects (web_server _CONFIG_FIELD_META) and the desktop settings
@@ -776,8 +648,7 @@ STT_MODEL_CATALOG = {
"local": ["base", "tiny", "small", "medium", "large-v3"],
"groq": ["whisper-large-v3-turbo", "whisper-large-v3", "distil-whisper-large-v3-en"],
"openai": ["whisper-1", "gpt-4o-mini-transcribe", "gpt-4o-transcribe", "gpt-transcribe"],
"elevenlabs": ["scribe_v2", "scribe_v1"],
}
"elevenlabs": ["scribe_v2", "scribe_v1"]}
# ElevenLabs historically uses ``model_id`` instead of ``model``.
_STT_MODEL_CONFIG_KEY = {"elevenlabs": "model_id"}
@@ -790,44 +661,36 @@ def _configure_stt_model(stt_provider: str, config: dict) -> None:
catalog = STT_MODEL_CATALOG.get(stt_provider)
if not catalog:
return
stt_cfg = _cfg_section(config, "stt")
prov_cfg = _cfg_section(stt_cfg, stt_provider)
prov_cfg = _cfg_section(_cfg_section(config, "stt"), stt_provider)
model_key = _STT_MODEL_CONFIG_KEY.get(stt_provider, "model")
current = str(prov_cfg.get(model_key) or "").strip()
ordered = list(catalog)
default_idx = ordered.index(current) if current in ordered else 0
idx = _prompt_choice(" Select STT model:", ordered, default_idx)
chosen = ordered[idx]
chosen = ordered[_prompt_choice(" Select STT model:", ordered, default_idx)]
prov_cfg[model_key] = chosen
_print_success(f" STT model set to: {chosen}")
# Provider-row marker key -> config section it selects into.
_PROVIDER_MARKER_SECTIONS = {
"tts_provider": "tts",
"stt_provider": "stt",
"browser_provider": "browser",
"web_backend": "web",
"image_gen_plugin_name": "image_gen",
"imagegen_backend": "image_gen",
"video_gen_plugin_name": "video_gen",
"tts_provider": "tts", "stt_provider": "stt", "browser_provider": "browser", "web_backend": "web",
"image_gen_plugin_name": "image_gen", "imagegen_backend": "image_gen", "video_gen_plugin_name": "video_gen",
}
def _drop_use_gateway(section) -> None:
if isinstance(section, dict):
section.pop("use_gateway", None)
def _write_provider_config(provider: dict, config: dict, *, managed_feature) -> None:
"""Persist the provider/backend config keys for a selected provider.
Pure, non-interactive core of :func:`_configure_provider` (no env prompts, post-setup hooks, Nous auth
gating or model pickers) shared by the CLI and the GUI ``PUT .../provider`` endpoint. Each pick writes
exactly ONE provider string per category (``nous`` for managed rows) and removes any legacy
``use_gateway`` key so the read-time shim cannot override the new choice.
"""
Pure, non-interactive core of :func:`_configure_provider` (no env prompts, post-setup hooks, Nous
auth gating or model pickers) shared by the CLI and the GUI ``PUT .../provider`` endpoint. Each pick
writes exactly ONE provider string per category (``nous`` for managed rows) and removes any legacy
``use_gateway`` key so the read-time shim cannot override the new choice."""
from hermes_cli.tools_config import TOOL_CATEGORIES, _cfg_section
def _drop_use_gateway(section) -> None:
if isinstance(section, dict):
section.pop("use_gateway", None)
def _set_selection(section_key: str, name_key: str, vendor_value) -> None:
section = _cfg_section(config, section_key)
section[name_key] = NOUS_MANAGED_PROVIDER if managed_feature else vendor_value
@@ -909,8 +772,7 @@ def apply_provider_selection(ts_key: str, provider_name: str, config: dict) -> N
selections = [
("image_gen", provider.get("image_gen_plugin_name")),
("video_gen", provider.get("video_gen_plugin_name")),
("image_gen", "fal" if provider.get("imagegen_backend") and not managed_feature else None),
]
("image_gen", "fal" if provider.get("imagegen_backend") and not managed_feature else None)]
for section_key, vendor in selections:
if vendor:
section = _cfg_section(config, section_key)
@@ -920,11 +782,10 @@ def apply_provider_selection(ts_key: str, provider_name: str, config: dict) -> N
def _nous_provider_gate(provider: dict, config: dict, managed_feature, *, force_fresh: bool) -> bool:
"""Return False (after printing why) when a Nous-gated row cannot be selected.
Managed Tool Gateway rows are always listed but only *activate* with paid Nous Portal access — selecting
one runs an inline Portal login (auth + entitlement only, no inference-provider switch). Pure pre-auth UX
rows (``requires_nous_auth`` without a managed feature) keep the older logged-in + entitled gate.
"""
Managed Tool Gateway rows are always listed but only *activate* with paid Nous Portal access —
selecting one runs an inline Portal login (auth + entitlement only, no inference-provider switch).
Pure pre-auth UX rows (``requires_nous_auth`` without a managed feature) keep the older logged-in +
entitled gate."""
from hermes_cli.tools_config import get_nous_subscription_features
if managed_feature:
@@ -932,8 +793,7 @@ def _nous_provider_gate(provider: dict, config: dict, managed_feature, *, force_
if not ensure_nous_portal_access(
capability=f"{provider.get('name', 'the Nous Tool Gateway')}",
coverage_category=MANAGED_FEATURE_COVERAGE_CATEGORY.get(managed_feature),
):
coverage_category=MANAGED_FEATURE_COVERAGE_CATEGORY.get(managed_feature)):
_print_warning(" Not enabled — Nous Portal access is required for this backend.")
return False
return True
@@ -943,8 +803,7 @@ def _nous_provider_gate(provider: dict, config: dict, managed_feature, *, force_
entitled = bool(features.account_info and features.account_info.paid_service_access is True)
if not features.nous_auth_present or not entitled:
message = format_nous_portal_entitlement_message(
features.account_info, capability=f"{provider.get('name', 'Nous Subscription')}"
)
features.account_info, capability=f"{provider.get('name', 'Nous Subscription')}")
_print_warning(f" {message or 'Nous Subscription is only available after logging into Nous Portal.'}")
return False
return True
@@ -994,10 +853,8 @@ def _print_provider_selection(provider: dict, managed_feature, *, reconfigure: b
if provider.get("browser_engine") and provider["browser_engine"] != "auto":
_print_success(f" Browser engine set to: {provider['browser_engine']}")
if provider.get("web_backend"):
if reconfigure and provider.get("web_tier"):
_print_success(f" Web backend set to: {provider['web_backend']} ({provider['web_tier']} tier)")
else:
_print_success(f" Web backend set to: {provider['web_backend']}")
tier = f" ({provider['web_tier']} tier)" if reconfigure and provider.get("web_tier") else ""
_print_success(f" Web backend set to: {provider['web_backend']}{tier}")
if reconfigure and provider.get("computer_use_backend"):
_print_success(f" Computer Use backend set to: {provider['computer_use_backend']}")
@@ -1019,61 +876,58 @@ def _show_portal_hint(provider: dict, config: dict, managed_feature, force_fresh
return False
def _prompt_secret(
key: str, label: str, url: str, default_val: str, *, reconfigure: bool, url_label: str, strip: bool = False,
) -> bool:
"""One env-var prompt; True unless the new-enable flow skipped the key.
Reconfigure mode shows the current value and re-prompts ("Enter to keep current"); the new-enable
flow prompts with ``default_val`` visible when one exists, else as a password, and ``strip`` decides
whether whitespace is trimmed (and whitespace-only counts as skipped)."""
if reconfigure:
existing = get_env_value(key)
if existing:
_print_info(f" {key}: configured ({existing[:8]}...)")
if url:
_print_info(f" {url_label}: {url}")
if reconfigure:
value = _prompt(f" {label} (Enter to keep current)", password=not default_val)
if value and value.strip():
save_env_value(key, value.strip())
_print_success(" Updated")
else:
_print_info(" Kept current")
return True
value = _prompt(f" {label}", default_val) if default_val else _prompt(f" {label}", password=True)
if strip:
value = (value or "").strip()
if value:
save_env_value(key, value)
_print_success(" Saved")
return True
_print_warning(" Skipped")
return False
def _prompt_env_vars(env_vars: list, *, reconfigure: bool) -> bool:
"""Prompt for a provider's env vars; True when every key ended up configured.
Reconfigure mode re-prompts every key ("Enter to keep current") and always returns True; the new-enable
flow keeps already-set keys without asking and reports False on any skipped key.
"""
Reconfigure mode re-prompts every key and always returns True; the new-enable flow keeps already-set
keys without asking and reports False on any skipped key."""
all_configured = True
for var in env_vars:
existing = get_env_value(var["key"])
if reconfigure:
if existing:
_print_info(f" {var['key']}: configured ({existing[:8]}...)")
url = var.get("url", "")
if url:
_print_info(f" Get yours at: {url}")
default_val = var.get("default", "")
value = _prompt(f" {var.get('prompt', var['key'])} (Enter to keep current)", password=not default_val)
if value and value.strip():
save_env_value(var["key"], value.strip())
_print_success(" Updated")
else:
_print_info(" Kept current")
continue
if existing:
if not reconfigure and get_env_value(var["key"]):
_print_success(f" {var['key']}: already configured")
continue
url = var.get("url", "")
if url:
_print_info(f" Get yours at: {url}")
default_val = var.get("default", "")
if default_val:
value = _prompt(f" {var.get('prompt', var['key'])}", default_val)
else:
value = _prompt(f" {var.get('prompt', var['key'])}", password=True)
if value:
save_env_value(var["key"], value)
_print_success(" Saved")
else:
_print_warning(" Skipped")
all_configured = False
ok = _prompt_secret(var["key"], var.get("prompt", var["key"]), var.get("url", ""), var.get("default", ""),
reconfigure=reconfigure, url_label="Get yours at")
all_configured = all_configured and ok
return all_configured
def _configure_provider(
provider: dict,
config: dict,
*,
force_fresh: bool = True,
reconfigure: bool = False,
):
def _configure_provider(provider: dict, config: dict, *, force_fresh: bool = True, reconfigure: bool = False):
"""Configure a single provider - prompt for API keys and set config.
``reconfigure=False`` (new-enable): already-set keys are kept without asking and the post-setup hook only
runs when every key was provided. ``reconfigure=True`` re-prompts every key and always runs the hook.
"""
``reconfigure=False`` (new-enable): already-set keys are kept without asking and the post-setup hook
only runs when every key was provided. ``reconfigure=True`` re-prompts every key and always runs the
hook."""
from hermes_cli.tools_config import _run_post_setup
env_vars = provider.get("env_vars", [])
@@ -1114,10 +968,9 @@ def _reconfigure_provider(provider: dict, config: dict, *, force_fresh: bool = T
def _configure_vision_backend() -> None:
"""Interactive vision-backend configuration (``auxiliary.vision.{provider,model,base_url}``).
Offers any authenticated provider + model (same surface as ``hermes model``) or a custom endpoint rather
than forcing OpenRouter. "Auto" leaves the keys empty so the resolver uses the main-model fallback chain.
"""
Offers any authenticated provider + model (same surface as ``hermes model``) or a custom endpoint
rather than forcing OpenRouter. "Auto" leaves the keys empty so the resolver uses the main-model
fallback chain."""
from hermes_cli.tools_config import _cfg_section, _prompt_choice
print()
@@ -1128,8 +981,7 @@ def _configure_vision_backend() -> None:
"Auto — use your main model / aggregator fallback (recommended)",
"Pick a provider and model",
"Custom OpenAI-compatible endpoint — base URL, API key, model",
"Skip",
]
"Skip"]
idx = _prompt_choice(" Configure vision backend", choices, 0)
config = load_config()
@@ -1141,13 +993,9 @@ def _configure_vision_backend() -> None:
vision_cfg.pop(key, None)
save_config(config)
_print_success(" Vision set to auto (main model / aggregator fallback)")
return
if idx == 1:
elif idx == 1:
_configure_vision_provider_model(config, vision_cfg)
return
if idx == 2:
elif idx == 2:
base_url = _prompt(" Base URL (blank for OpenAI)").strip() or "https://api.openai.com/v1"
is_native_openai = base_url_hostname(base_url) == "api.openai.com"
key_label = " OPENAI_API_KEY" if is_native_openai else " API key"
@@ -1169,18 +1017,15 @@ def _configure_vision_backend() -> None:
vision_cfg.pop("model", None)
save_config(config)
_print_success(f" Vision set to custom endpoint{f' ({model})' if model else ''}")
return
_print_info(" Skipped vision configuration")
else:
_print_info(" Skipped vision configuration")
def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None:
"""Provider + model picker for vision, mirroring the ``/model`` surface.
Rows come from ``build_aux_picker_rows()`` so this lists exactly what the ``hermes model`` aux-task picker
lists, including user-defined ``providers:`` / ``custom_providers:`` endpoints. Persists
``auxiliary.vision.provider`` + ``.model``.
"""
Rows come from ``build_aux_picker_rows()`` so this lists exactly what the ``hermes model`` aux-task
picker lists, including user-defined ``providers:`` / ``custom_providers:`` endpoints. Persists
``auxiliary.vision.provider`` + ``.model``."""
from hermes_cli.tools_config import _prompt_choice
try:
@@ -1203,8 +1048,7 @@ def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None:
if not providers:
_print_warning(
" No authenticated providers found. Configure a provider first "
"with `hermes model`, then re-run this."
)
"with `hermes model`, then re-run this.")
return
provider_labels = [label for _slug, label, _models in format_aux_picker_entries(
@@ -1240,10 +1084,8 @@ def _configure_vision_provider_model(config: dict, vision_cfg: dict) -> None:
def _configure_simple_requirements(ts_key: str, *, reconfigure: bool = False):
"""Fallback for toolsets that just need env vars (no provider selection).
Vision has its own provider/model picker — run it directly so neither flow falls back to the generic
single-key prompt (which would re-ask for OPENROUTER_API_KEY).
"""
single-key prompt (which would re-ask for OPENROUTER_API_KEY)."""
from hermes_cli.tools_config import TOOLSET_ENV_REQUIREMENTS, _toolset_has_keys, _toolset_label
if ts_key == "vision":
@@ -1265,18 +1107,4 @@ def _configure_simple_requirements(ts_key: str, *, reconfigure: bool = False):
print(color(f" {ts_label} requires configuration:", Colors.YELLOW))
for var, url in requirements:
if reconfigure:
existing = get_env_value(var)
if existing:
_print_info(f" {var}: configured ({existing[:8]}...)")
if url:
_print_info(f" Get key at: {url}")
suffix = " (Enter to keep current)" if reconfigure else ""
value = _prompt(f" {var}{suffix}", password=True)
if value and value.strip():
save_env_value(var, value.strip())
_print_success(" Updated" if reconfigure else " Saved")
elif reconfigure:
_print_info(" Kept current")
else:
_print_warning(" Skipped")
_prompt_secret(var, var, url, "", reconfigure=reconfigure, url_label="Get key at", strip=True)
hermes_cli/toolset_scope.py
+1 -4
View File
@@ -4,10 +4,7 @@ from typing import Set
# Toolsets without a restriction entry are available on every platform.
_TOOLSET_PLATFORM_RESTRICTIONS = {
"discord": {"discord"},
"discord_admin": {"discord"},
}
_TOOLSET_PLATFORM_RESTRICTIONS = {"discord": {"discord"}, "discord_admin": {"discord"}}
def toolset_allowed_for_platform(ts_key: str, platform: str) -> bool:
hermes_cli/toolset_validation.py
+25 -45
View File
@@ -14,17 +14,12 @@ def _platform_default_toolset(platform: object) -> str:
def _platform_default_is_valid(
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)
):
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)):
return True
# Dynamic plugin platforms are resolved by toolsets.resolve_toolset() even
# though their synthesized hermes-<platform> name is not in TOOLSETS.
# Dynamic plugin platforms are resolved by toolsets.resolve_toolset() even though their synthesized
# hermes-<platform> name is not in TOOLSETS.
try:
from gateway.platform_registry import platform_registry
@@ -34,19 +29,15 @@ 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.
Reports: a toolset name ``is_valid_toolset`` rejects (suggesting ``hermes-<platform>`` when
that would have been valid); a non-empty mapping resolving to zero valid toolsets (agent
would start with no tools); a platform with no valid toolsets, checked per-platform because
the global net is suppressed once any platform is valid; and non-list platform values, which
fall back to the platform default. ``is_valid_toolset`` is injected so this does no registry
imports or I/O.
"""
Reports: a toolset name ``is_valid_toolset`` rejects (suggesting ``hermes-<platform>`` when that
would have been valid); a non-empty mapping resolving to zero valid toolsets (agent would start with
no tools); a platform with no valid toolsets, checked per-platform because the global net is
suppressed once any platform is valid; and non-list platform values, which fall back to the platform
default. ``is_valid_toolset`` is injected so this does no registry imports or I/O."""
warnings: List[str] = []
if not isinstance(platform_toolsets, dict) or not platform_toolsets:
return warnings
@@ -54,17 +45,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):
@@ -73,8 +60,7 @@ def validate_platform_toolsets(
value_detail = f"invalid {type(raw).__name__} toolset value"
warnings.append(
f"platform '{platform}' has {value_detail} — "
f"{fallback_detail}. Run `hermes tools` to configure explicitly."
)
f"{fallback_detail}. Run `hermes tools` to configure explicitly.")
if platform_valid_count == 0:
warnings.append(f"platform '{platform}' has no valid toolsets configured — {_NO_TOOLS}")
continue
@@ -82,21 +68,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 "
f"'{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"
@@ -105,6 +86,5 @@ def validate_platform_toolsets(
if valid_count == 0:
warnings.append(
"platform_toolsets resolves to zero valid toolsets — the agent will "
"have no tools. Run `hermes tools` to reconfigure."
)
"have no tools. Run `hermes tools` to reconfigure.")
return warnings