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

Merge branch 'simp/r3-23-g4' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:59:08 -07:00
parent a3cd925274 91d144f770
commit 6f44752bc4
8 changed files with 628 additions and 1000 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/secret_prompt.py
+14 -40
View File
@@ -10,15 +10,11 @@ from collections.abc import Callable
_BACKSPACE_CHARS = {"\b", "\x7f"}
_ENTER_CHARS = {"\r", "\n"}
_EOF_CHARS = {"\x04", "\x1a"}
_EOF_CHARS = {"\x04", "\x1a", ""} # "" == stream closed
def _collect_masked_input(
read_char: Callable[[], str],
write: Callable[[str], object],
prompt: str,
*,
mask: str = "*",
read_char: Callable[[], str], write: Callable[[str], object], prompt: str, *, mask: str = "*",
) -> str:
"""Read one secret line while writing a mask character per typed char."""
value: list[str] = []
@@ -26,9 +22,6 @@ def _collect_masked_input(
while True:
ch = read_char()
if ch == "":
write("\r\n")
raise EOFError
if ch in _ENTER_CHARS:
write("\r\n")
return "".join(value)
@@ -44,10 +37,9 @@ def _collect_masked_input(
write("\b \b")
continue
if ch == "\x1b":
# Ignore escape itself. Terminals commonly send escape-prefixed
# navigation/delete sequences; they should not become secret text.
# Terminals send escape-prefixed navigation/delete sequences; they must not become
# secret text.
continue
value.append(ch)
if mask:
write(mask)
@@ -59,22 +51,12 @@ def masked_secret_prompt(prompt: str, *, mask: str = "*") -> str:
Falls back to ``getpass.getpass`` when stdin/stdout are not interactive or when raw terminal
handling is unavailable.
"""
stdin = sys.stdin
stdout = sys.stdout
if not _stream_is_tty(stdin) or not _stream_is_tty(stdout):
if not _stream_is_tty(sys.stdin) or not _stream_is_tty(sys.stdout):
return getpass.getpass(prompt)
if os.name == "nt":
try:
return _masked_secret_prompt_windows(prompt, mask=mask)
except (KeyboardInterrupt, EOFError):
raise
except Exception:
return getpass.getpass(prompt)
masked = _masked_secret_prompt_windows if os.name == "nt" else _masked_secret_prompt_posix
try:
return _masked_secret_prompt_posix(prompt, mask=mask)
return masked(prompt, mask=mask)
except (KeyboardInterrupt, EOFError):
raise
except Exception:
@@ -88,6 +70,11 @@ def _stream_is_tty(stream) -> bool:
return False
def _write(text: str) -> None:
sys.stdout.write(text)
sys.stdout.flush()
def _masked_secret_prompt_windows(prompt: str, *, mask: str) -> str:
import msvcrt
@@ -98,29 +85,16 @@ def _masked_secret_prompt_windows(prompt: str, *, mask: str) -> str:
return "\x1b"
return ch
def write(text: str) -> None:
sys.stdout.write(text)
sys.stdout.flush()
return _collect_masked_input(read_char, write, prompt, mask=mask)
return _collect_masked_input(read_char, _write, prompt, mask=mask)
def _masked_secret_prompt_posix(prompt: str, *, mask: str) -> str:
import termios
import tty
fd = sys.stdin.fileno()
old_attrs = termios.tcgetattr(fd)
def read_char() -> str:
return sys.stdin.read(1)
def write(text: str) -> None:
sys.stdout.write(text)
sys.stdout.flush()
try:
tty.setraw(fd)
return _collect_masked_input(read_char, write, prompt, mask=mask)
return _collect_masked_input(lambda: sys.stdin.read(1), _write, prompt, mask=mask)
finally:
termios.tcsetattr(fd, termios.TCSADRAIN, old_attrs)
hermes_cli/secrets_cli.py
+176 -282
View File
@@ -15,49 +15,26 @@ from rich.console import Console
from rich.panel import Panel
from rich.table import Table
# NOTE: the Bitwarden backend (``agent.secret_sources.bitwarden``) pulls in
# ``cryptography`` at module-import time. On Windows the resulting
# ``cryptography._rust.pyd`` is mapped into the running process — and when
# that process is ``hermes update``, the self-lock preflight detects the
# loaded native module and defers (#86781). Keep the backend import lazy:
# this module is registered parse-time from ``hermes_cli.main`` and must not
# touch ``bw`` until a handler actually runs.
#
# ``_BWS_VERSION`` is duplicated here (as a plain string) so ``register_cli``
# can render the ``install --help`` text without importing the backend.
# ``agent.secret_sources.bitwarden._BWS_VERSION`` is the source of truth;
# bump both together when pinning a new bws release.
# The Bitwarden backend pulls in ``cryptography`` at import time; on Windows that mapped native
# module makes the ``hermes update`` self-lock preflight defer. This module is registered
# parse-time from ``hermes_cli.main``, so the backend import stays lazy (nothing touches ``bw``
# until a handler runs) and ``_BWS_VERSION`` is duplicated here for the ``install --help`` text.
# ``agent.secret_sources.bitwarden._BWS_VERSION`` is the source of truth; bump both together.
_BWS_VERSION = "2.0.0"
from hermes_cli._secrets_common import (
arg,
cfg_str,
cli_version,
disable_secret_source,
flag,
print_status_panel,
print_table,
prompt_index,
register_subcommands,
require_enabled,
rotate_token,
secret_cli_env,
section_cfg,
arg, cfg_str, cli_version, disable_secret_source, flag, print_status_panel, print_table,
prompt_index, register_subcommands, require_enabled, rotate_token, secret_cli_env, section_cfg,
yn,
)
from hermes_cli.config import (
get_env_path,
load_config,
save_config,
save_env_value,
)
from hermes_cli.config import get_env_path, load_config, save_config, save_env_value
from hermes_cli.secret_prompt import masked_secret_prompt
# Old names kept bound: tests monkeypatch ``secrets_cli._bws_version``.
_bws_version = cli_version
_yn = yn
_DEFAULT_TOKEN_ENV = "BWS_ACCESS_TOKEN"
_NOT_BSM_TOKEN_WARNING = (
"[yellow]Warning: token doesn't start with '0.' — usually that means "
"you pasted something other than a BSM access token.[/yellow]"
@@ -71,17 +48,12 @@ _NOT_BSM_TOKEN_WARNING_CONTINUING = (
def _load_bw():
"""Import ``agent.secret_sources.bitwarden`` on first use (crypto payload)."""
from agent.secret_sources import bitwarden as _bw
return _bw
def __getattr__(name: str):
"""PEP 562 module-level lazy resolver.
Existing callers (and upstream tests) monkeypatch attributes on ``hermes_cli.secrets_cli.bw``
directly. Resolving that attribute at module-import time would re-import ``cryptography``
eagerly — the very self-lock we are preventing (#86781).
"""
"""PEP 562 lazy ``bw`` attribute (tests monkeypatch ``secrets_cli.bw``); an eager binding
would re-import ``cryptography`` at import time."""
if name == "bw":
return _load_bw()
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
@@ -125,47 +97,85 @@ def register_cli(parent_parser: argparse.ArgumentParser) -> None:
# ---------------------------------------------------------------------------
def cmd_setup(args: argparse.Namespace) -> int:
bw = _load_bw()
console = Console()
console.print(
Panel.fit(
"[bold]Bitwarden Secrets Manager setup[/bold]\n\n"
"Need an access token? In the Bitwarden web app:\n"
" Secrets Manager → Machine accounts → [your account] →\n"
" Access tokens → Create access token\n\n"
"Copy the token (starts with [cyan]0.[/cyan]…) — it cannot be retrieved later.",
border_style="cyan",
)
)
# ------------------------------------------------------------------ binary
def _step(console: Console, n: int, title: str) -> None:
console.print()
console.print("[bold]Step 1[/bold] Install the bws CLI")
console.print(f"[bold]Step {n}[/bold] {title}")
def _setup_binary(bw, console: Console) -> Optional[Path]:
"""Step 1: locate or download bws; None (after printing) on failure."""
_step(console, 1, "Install the bws CLI")
try:
binary = bw.find_bws(install_if_missing=False)
if binary is None:
console.print(" No bws on PATH — downloading…")
binary = bw.install_bws()
version = _bws_version(binary)
console.print(f" [green]✓[/green] {binary} ({version})")
console.print(f" [green]✓[/green] {binary} ({_bws_version(binary)})")
return binary
except Exception as exc: # noqa: BLE001
console.print(f" [red]✗ Could not install bws: {exc}[/red]")
console.print(
" Manual install: "
"https://github.com/bitwarden/sdk-sm/releases"
)
console.print(" Manual install: https://github.com/bitwarden/sdk-sm/releases")
return None
def _missing_noninteractive_flags(args: argparse.Namespace) -> list[str]:
"""Setup flags a no-TTY run must supply (BWS_SERVER_URL env substitutes for --server-url)."""
provided = {
"--access-token": args.access_token,
"--server-url": (args.server_url or "").strip() or os.environ.get("BWS_SERVER_URL", ""),
"--project-id": args.project_id}
return [flag for flag, value in provided.items() if not (value and value.strip())]
def _setup_token(args: argparse.Namespace, console: Console, token_env: str) -> Optional[str]:
"""Step 2: take the token from ``--access-token`` or a masked prompt and persist it."""
_step(console, 2, "Provide your access token")
token = (args.access_token or "").strip() or masked_secret_prompt(f" Paste access token ({token_env}): ").strip()
if not token:
console.print(" [red]Empty token, aborting.[/red]")
return None
if not token.startswith("0."):
console.print(_NOT_BSM_TOKEN_WARNING_CONTINUING)
save_env_value(token_env, token)
os.environ[token_env] = token # so the test fetch below sees it
console.print(f" [green]✓[/green] stored in {get_env_path()} as {token_env}")
return token
def _setup_project(binary: Path, token: str, console: Console, server_url: str) -> Optional[str]:
"""Step 4: list projects and let the user pick one; None (after printing) when none usable."""
_step(console, 4, "Pick a project")
projects = _list_projects(binary, token, console, server_url=server_url)
if projects is None:
return None
if not projects:
console.print(" [yellow]No projects visible to this machine account.[/yellow]")
console.print(" In the Bitwarden web app, open the machine account → Projects tab "
"and grant it access to at least one project.")
return None
print_table(console, (("#", {"style": "cyan", "width": 4}), "Name", ("ID", {"style": "dim"})),
((str(i), p.get("name", "?"), p.get("id", "?")) for i, p in enumerate(projects, 1)))
idx = prompt_index(console, f" Select project [1-{len(projects)}]: ", len(projects))
return projects[idx - 1]["id"]
def cmd_setup(args: argparse.Namespace) -> int:
bw = _load_bw()
console = Console()
console.print(Panel.fit(
"[bold]Bitwarden Secrets Manager setup[/bold]\n\n"
"Need an access token? In the Bitwarden web app:\n"
" Secrets Manager → Machine accounts → [your account] →\n"
" Access tokens → Create access token\n\n"
"Copy the token (starts with [cyan]0.[/cyan]…) — it cannot be retrieved later.",
border_style="cyan"))
binary = _setup_binary(bw, console)
if binary is None:
return 1
# -- non-interactive guard --
if not sys.stdin.isatty():
# BWS_SERVER_URL env var is accepted as a non-interactive substitute for --server-url.
provided = {
"--access-token": args.access_token,
"--server-url": (args.server_url or "").strip() or os.environ.get("BWS_SERVER_URL", ""),
"--project-id": args.project_id,
}
missing = [flag for flag, value in provided.items() if not (value and value.strip())]
missing = _missing_noninteractive_flags(args)
if missing:
console.print(
f" [red]Non-interactive mode (no TTY) requires all setup flags.[/red]\n"
@@ -174,79 +184,34 @@ def cmd_setup(args: argparse.Namespace) -> int:
" hermes secrets bitwarden setup \\\n"
" --access-token '0.xxx' \\\n"
" --server-url 'https://vault.bitwarden.com' \\\n"
" --project-id 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'"
)
" --project-id 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx'")
return 1
# ------------------------------------------------------------------- token
console.print()
console.print("[bold]Step 2[/bold] Provide your access token")
cfg = load_config()
secrets_cfg = cfg.setdefault("secrets", {}).setdefault("bitwarden", {})
token_env = secrets_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
token = (args.access_token or "").strip()
if not token:
token = masked_secret_prompt(f" Paste access token ({token_env}): ").strip()
if not token:
console.print(" [red]Empty token, aborting.[/red]")
token_env = secrets_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV)
token = _setup_token(args, console, token_env)
if token is None:
return 1
if not token.startswith("0."):
console.print(_NOT_BSM_TOKEN_WARNING_CONTINUING)
save_env_value(token_env, token)
os.environ[token_env] = token # so the test fetch below sees it
console.print(f" [green]✓[/green] stored in {get_env_path()} as {token_env}")
# ------------------------------------------------------------------ region
console.print()
console.print("[bold]Step 3[/bold] Pick a Bitwarden region")
_step(console, 3, "Pick a Bitwarden region")
server_url = _resolve_server_url(args, secrets_cfg, console)
if server_url is None:
return 1
if server_url:
console.print(f" [green]✓[/green] using {server_url}")
else:
console.print(
" [green]✓[/green] using bws default "
"(US Cloud, https://vault.bitwarden.com)"
)
console.print(f" [green]✓[/green] using {server_url}" if server_url
else " [green]✓[/green] using bws default (US Cloud, https://vault.bitwarden.com)")
# ------------------------------------------------------------------- project
project_given = bool(args.project_id and args.project_id.strip())
if project_given:
project_id = args.project_id.strip()
else:
console.print()
console.print("[bold]Step 4[/bold] Pick a project")
projects = _list_projects(binary, token, console, server_url=server_url)
if projects is None:
return 1
if not projects:
console.print(" [yellow]No projects visible to this machine account.[/yellow]")
console.print(
" In the Bitwarden web app, open the machine account → Projects tab "
"and grant it access to at least one project."
)
project_id = (args.project_id or "").strip()
project_given = bool(project_id)
if not project_given:
project_id = _setup_project(binary, token, console, server_url)
if project_id is None:
return 1
print_table(console, (("#", {"style": "cyan", "width": 4}), "Name", ("ID", {"style": "dim"})),
((str(i), p.get("name", "?"), p.get("id", "?")) for i, p in enumerate(projects, 1)))
idx = prompt_index(console, f" Select project [1-{len(projects)}]: ", len(projects))
project_id = projects[idx - 1]["id"]
# ------------------------------------------------------------------- test
console.print()
console.print(f"[bold]Step {4 if project_given else 5}[/bold] Test fetch")
_step(console, 4 if project_given else 5, "Test fetch")
try:
secrets, warnings = bw.fetch_bitwarden_secrets(
access_token=token,
project_id=project_id,
binary=binary,
use_cache=False,
server_url=server_url,
)
access_token=token, project_id=project_id, binary=binary, use_cache=False, server_url=server_url)
except Exception as exc: # noqa: BLE001
console.print(f" [red]✗ Fetch failed: {exc}[/red]")
return 1
@@ -259,26 +224,18 @@ def cmd_setup(args: argparse.Namespace) -> int:
for w in warnings:
console.print(f" [yellow]warning:[/yellow] {w}")
# ------------------------------------------------------------------- save
secrets_cfg["enabled"] = True
secrets_cfg["project_id"] = project_id
secrets_cfg["server_url"] = server_url
secrets_cfg.setdefault("access_token_env", token_env)
secrets_cfg.setdefault("cache_ttl_seconds", 300)
secrets_cfg.setdefault("override_existing", True)
secrets_cfg.setdefault("auto_install", True)
secrets_cfg.update(enabled=True, project_id=project_id, server_url=server_url)
for key, default in (("access_token_env", token_env), ("cache_ttl_seconds", 300),
("override_existing", True), ("auto_install", True)):
secrets_cfg.setdefault(key, default)
save_config(cfg)
console.print()
console.print(
"[green]✓ Bitwarden Secrets Manager is enabled.[/green] "
"Secrets will be pulled at the start of every Hermes process."
)
console.print(
" Status: [cyan]hermes secrets bitwarden status[/cyan]\n"
" Refresh: [cyan]hermes secrets bitwarden sync[/cyan]\n"
" Disable: [cyan]hermes secrets bitwarden disable[/cyan]"
)
console.print("[green]✓ Bitwarden Secrets Manager is enabled.[/green] "
"Secrets will be pulled at the start of every Hermes process.")
console.print(" Status: [cyan]hermes secrets bitwarden status[/cyan]\n"
" Refresh: [cyan]hermes secrets bitwarden sync[/cyan]\n"
" Disable: [cyan]hermes secrets bitwarden disable[/cyan]")
return 0
@@ -300,31 +257,25 @@ def cmd_status(args: argparse.Namespace) -> int:
bw_cfg = _bw_cfg(load_config())
enabled = bool(bw_cfg.get("enabled"))
token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
token_env = bw_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV)
project_id = bw_cfg.get("project_id", "")
server_url = cfg_str(bw_cfg, "server_url")
token = os.environ.get(token_env, "").strip()
token_set = bool(token)
binary = bw.find_bws(install_if_missing=False)
token_validation, validation_messages = _token_validation_status(
enabled=enabled,
binary=binary,
token=token,
server_url=server_url,
)
enabled=enabled, binary=binary, token=token, server_url=server_url)
print_status_panel(console, "Bitwarden Secrets Manager", (
("Enabled", _yn(enabled)),
("Token env var", token_env),
("Token in env", _yn(token_set)),
("Token in env", _yn(bool(token))),
("Token validation", token_validation),
("Project ID", project_id or "[dim](unset)[/dim]"),
("Server URL", server_url or "[dim]default (US Cloud, https://vault.bitwarden.com)[/dim]"),
("Override existing", _yn(bool(bw_cfg.get("override_existing", False)))),
("Cache TTL (s)", str(bw_cfg.get("cache_ttl_seconds", 300))),
("Auto-install", _yn(bool(bw_cfg.get("auto_install", True)))),
("bws binary",
f"{binary} ({_bws_version(binary)})" if binary else "[yellow]not installed[/yellow]"),
("bws binary", f"{binary} ({_bws_version(binary)})" if binary else "[yellow]not installed[/yellow]"),
))
for message in validation_messages:
console.print(message)
@@ -332,29 +283,22 @@ def cmd_status(args: argparse.Namespace) -> int:
if not enabled:
console.print("\n Run [cyan]hermes secrets bitwarden setup[/cyan] to enable.")
return 0
if not token_set:
console.print(
f"\n [yellow]Enabled but {token_env} is not set — Hermes will skip BSM "
"and warn on next startup.[/yellow]"
)
if not token:
console.print(f"\n [yellow]Enabled but {token_env} is not set — Hermes will skip BSM "
"and warn on next startup.[/yellow]")
if not project_id:
console.print(
"\n [yellow]Enabled but no project_id — nothing to fetch.[/yellow]"
)
console.print("\n [yellow]Enabled but no project_id — nothing to fetch.[/yellow]")
return 0
def cmd_token(args: argparse.Namespace) -> int:
"""Rotate the BSM access token without re-running the whole setup wizard.
Prompts for (or accepts via ``--access-token``) a new machine-account token, probes Bitwarden
with it (unless ``--no-verify``), and only then persists it to .env — so a bad paste never
bricks the working token.
"""
"""Rotate the BSM access token without re-running the whole setup wizard: probe Bitwarden
with the new token (unless ``--no-verify``) and only then persist it, so a bad paste never
bricks the working token."""
bw = _load_bw()
console = Console()
bw_cfg = _bw_cfg(load_config())
token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
token_env = bw_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV)
server_url = cfg_str(bw_cfg, "server_url")
def verify(token: str) -> bool:
@@ -364,30 +308,22 @@ def cmd_token(args: argparse.Namespace) -> int:
return True
binary = bw.find_bws(install_if_missing=True)
if binary is None:
console.print(
"[red]bws binary not available — cannot verify. "
"Re-run with --no-verify to store anyway.[/red]"
)
console.print("[red]bws binary not available — cannot verify. "
"Re-run with --no-verify to store anyway.[/red]")
return False
console.print("Verifying against Bitwarden…")
projects = _list_projects(binary, token, console, server_url=server_url)
if projects is None:
console.print(
"[red]✗ New token was rejected — nothing was changed.[/red]"
)
console.print("[red]✗ New token was rejected — nothing was changed.[/red]")
return False
console.print(
f"[green]✓ Token accepted[/green] "
f"({len(projects)} project{'s' if len(projects) != 1 else ''} visible)."
)
console.print(f"[green]✓ Token accepted[/green] "
f"({len(projects)} project{'s' if len(projects) != 1 else ''} visible).")
project_id = str(bw_cfg.get("project_id", "") or "")
if project_id and projects and project_id not in {p["id"] for p in projects}:
console.print(
f"[yellow]Warning: configured project {project_id} is not visible "
"to this machine account. Grant it access in the Bitwarden web "
"app or re-run `hermes secrets bitwarden setup` to pick a "
"different project.[/yellow]"
)
"app or re-run `hermes secrets bitwarden setup` to pick a different project.[/yellow]")
return True
return rotate_token(
@@ -405,8 +341,7 @@ def cmd_token(args: argparse.Namespace) -> int:
"[yellow]Note: the Bitwarden integration is currently disabled — "
"run `hermes secrets bitwarden setup` (or set "
"secrets.bitwarden.enabled: true) to turn it on.[/yellow]"
),
)
))
def cmd_sync(args: argparse.Namespace) -> int:
@@ -416,25 +351,19 @@ def cmd_sync(args: argparse.Namespace) -> int:
if not require_enabled(console, bw_cfg, "Bitwarden", "bitwarden"):
return 1
token_env = bw_cfg.get("access_token_env", "BWS_ACCESS_TOKEN")
token_env = bw_cfg.get("access_token_env", _DEFAULT_TOKEN_ENV)
token = os.environ.get(token_env, "").strip()
if not token:
console.print(f"[red]{token_env} is not set.[/red]")
return 1
project_id = bw_cfg.get("project_id", "")
if not project_id:
console.print("[red]No project_id configured.[/red]")
return 1
server_url = cfg_str(bw_cfg, "server_url")
try:
secrets, warnings = bw.fetch_bitwarden_secrets(
access_token=token,
project_id=project_id,
use_cache=False,
server_url=server_url,
access_token=token, project_id=project_id, use_cache=False, server_url=cfg_str(bw_cfg, "server_url"),
)
except Exception as exc: # noqa: BLE001
console.print(f"[red]Fetch failed: {exc}[/red]")
@@ -448,28 +377,25 @@ def cmd_sync(args: argparse.Namespace) -> int:
rows = []
applied = 0
for key in sorted(secrets):
if key == token_env:
rows.append((key, "[dim]skip (bootstrap token)[/dim]"))
continue
already = bool(os.environ.get(key))
if already and not override:
rows.append((key, "[dim]skip (already set)[/dim]"))
continue
if args.apply:
if key == token_env:
action = "[dim]skip (bootstrap token)[/dim]"
elif already and not override:
action = "[dim]skip (already set)[/dim]"
elif args.apply:
os.environ[key] = secrets[key]
applied += 1
rows.append((key, "[green]exported[/green]" + (" (overrode)" if already else "")))
action = "[green]exported[/green]" + (" (overrode)" if already else "")
else:
rows.append((key, "[green]would export[/green]" + (" (overrides)" if already else "")))
action = "[green]would export[/green]" + (" (overrides)" if already else "")
rows.append((key, action))
print_table(console, (("Name", {"style": "cyan"}), "Action"), rows, warnings)
if not args.apply:
console.print(
"\n This was a dry-run — secrets are picked up automatically on the "
"next [cyan]hermes[/cyan] invocation. Re-run with [cyan]--apply[/cyan] "
"to export into the current shell instead."
)
console.print("\n This was a dry-run — secrets are picked up automatically on the "
"next [cyan]hermes[/cyan] invocation. Re-run with [cyan]--apply[/cyan] "
"to export into the current shell instead.")
else:
console.print(f"\n [green]Exported {applied} secret(s) into current process.[/green]")
return 0
@@ -481,8 +407,7 @@ def cmd_disable(args: argparse.Namespace) -> int:
"[green]Disabled.[/green] Bitwarden secrets will NOT be pulled on the next "
"Hermes invocation.\n"
" Your access token is left in .env — remove it manually if you also want "
"to revoke the credential.",
)
"to revoke the credential.")
def cmd_install(args: argparse.Namespace) -> int:
@@ -503,27 +428,19 @@ def cmd_install(args: argparse.Namespace) -> int:
def _token_validation_status(
*,
enabled: bool,
binary: Optional[Path],
token: str,
server_url: str = "",
*, enabled: bool, binary: Optional[Path], token: str, server_url: str = "",
) -> tuple[str, list[str]]:
if not enabled:
return "[dim]not checked[/dim] (integration disabled)", []
if not token:
return "[dim]not checked[/dim] (token missing)", []
if binary is None:
return "[dim]not checked[/dim] (bws not installed)", []
for skipped, reason in ((not enabled, "integration disabled"), (not token, "token missing"),
(binary is None, "bws not installed")):
if skipped:
return f"[dim]not checked[/dim] ({reason})", []
messages: list[str] = []
if not token.startswith("0."):
messages.append(_NOT_BSM_TOKEN_WARNING_CONTINUING)
capture = io.StringIO()
probe_console = Console(file=capture, record=True, width=200)
projects = _list_projects(binary, token, probe_console, server_url=server_url)
if projects is None:
probe_console = Console(file=io.StringIO(), record=True, width=200)
if _list_projects(binary, token, probe_console, server_url=server_url) is None:
details = probe_console.export_text(styles=False).strip()
if details:
messages.extend(line.rstrip() for line in details.splitlines())
@@ -531,6 +448,20 @@ def _token_validation_status(
return "[green]passed[/green]", messages
# (substring of the lowercased bws error, follow-up hint) — first match wins.
_PROJECT_LIST_HINTS = (
(("invalid_client", "400 bad request"),
" [yellow]'invalid_client' from the US identity endpoint usually "
"means the token is for a different Bitwarden region. Re-run "
"[cyan]hermes secrets bitwarden setup[/cyan] and pick EU or "
"self-hosted at the region prompt, or set [cyan]secrets.bitwarden."
"server_url[/cyan] in config.yaml.[/yellow]"),
(("authorization", "invalid"),
" [yellow]This usually means the access token is wrong or revoked. "
"Double-check it in the Bitwarden web app.[/yellow]"),
)
def _list_projects(
binary: Path, token: str, console: Console, *, server_url: str = ""
) -> Optional[List[dict]]:
@@ -542,11 +473,7 @@ def _list_projects(
try:
res = subprocess.run(
[str(binary), "project", "list", "--output", "json"],
env=env,
capture_output=True,
text=True, encoding='utf-8', errors='replace',
timeout=15,
)
env=env, capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=15)
except (OSError, subprocess.TimeoutExpired) as exc:
console.print(f" [red]Couldn't list projects: {exc}[/red]")
return None
@@ -555,19 +482,10 @@ def _list_projects(
err = (res.stderr or res.stdout).strip()[:300]
console.print(f" [red]bws project list failed: {err}[/red]")
lowered = err.lower()
if "invalid_client" in lowered or "400 bad request" in lowered:
console.print(
" [yellow]'invalid_client' from the US identity endpoint usually "
"means the token is for a different Bitwarden region. Re-run "
"[cyan]hermes secrets bitwarden setup[/cyan] and pick EU or "
"self-hosted at the region prompt, or set [cyan]secrets.bitwarden."
"server_url[/cyan] in config.yaml.[/yellow]"
)
elif "authorization" in lowered or "invalid" in lowered:
console.print(
" [yellow]This usually means the access token is wrong or revoked. "
"Double-check it in the Bitwarden web app.[/yellow]"
)
for needles, hint in _PROJECT_LIST_HINTS:
if any(n in lowered for n in needles):
console.print(hint)
break
return None
try:
@@ -580,9 +498,7 @@ def _list_projects(
return [p for p in data if isinstance(p, dict) and p.get("id")]
# Canonical Bitwarden region endpoints. Keep in sync with what Bitwarden
# publishes — these are stable but if a third region appears, add it here
# and to the prompt below.
# Canonical Bitwarden region endpoints; add a new region here and it appears in the prompt.
_REGION_PRESETS = [
("US Cloud (https://vault.bitwarden.com — bws default)", ""),
("EU Cloud (https://vault.bitwarden.eu)", "https://vault.bitwarden.eu"),
@@ -590,66 +506,44 @@ _REGION_PRESETS = [
def _resolve_server_url(
args: argparse.Namespace,
secrets_cfg: dict,
console: Console,
args: argparse.Namespace, secrets_cfg: dict, console: Console,
) -> Optional[str]:
"""Pick a Bitwarden server URL for setup.
Resolution order: 1. ``--server-url`` CLI flag (non-interactive) 2. ``BWS_SERVER_URL`` env var
(so users running with that already set in their shell don't have to re-enter it) 3. Existing
``secrets.bitwarden.server_url`` value (for re-runs) 4. Interactive menu: US / EU / self-hosted
"""
"""Pick a Bitwarden server URL: ``--server-url``, then ``BWS_SERVER_URL``, then the existing
``secrets.bitwarden.server_url``, then the interactive US / EU / self-hosted menu. None (after
printing) when a custom URL is left empty."""
if args.server_url and args.server_url.strip():
return args.server_url.strip()
env_url = os.environ.get("BWS_SERVER_URL", "").strip()
if env_url:
console.print(
f" Detected [cyan]BWS_SERVER_URL[/cyan]={env_url} in your shell — using it."
)
console.print(f" Detected [cyan]BWS_SERVER_URL[/cyan]={env_url} in your shell — using it.")
return env_url
existing = cfg_str(secrets_cfg, "server_url")
if existing:
console.print(
f" Existing config: [cyan]{existing}[/cyan]. "
"Press Enter to keep, or pick a different option below."
)
console.print(f" Existing config: [cyan]{existing}[/cyan]. "
"Press Enter to keep, or pick a different option below.")
table = Table(show_header=True, header_style="bold", box=None, padding=(0, 2))
table.add_column("#", style="cyan", width=4)
table.add_column("Region / endpoint")
for i, (label, _url) in enumerate(_REGION_PRESETS, 1):
table.add_row(str(i), label)
table.add_row(str(len(_REGION_PRESETS) + 1), "Self-hosted / custom URL")
custom_idx = len(_REGION_PRESETS) + 1
table.add_row(str(custom_idx), "Self-hosted / custom URL")
console.print(table)
custom_idx = len(_REGION_PRESETS) + 1
prompt = f" Select region [1-{custom_idx}]"
if existing:
prompt += " (Enter to keep current)"
idx = prompt_index(
console,
prompt + ": ",
custom_idx,
allow_empty=bool(existing),
empty_message=" [red]Enter a number.[/red]",
)
prompt = f" Select region [1-{custom_idx}]" + (" (Enter to keep current)" if existing else "")
idx = prompt_index(console, prompt + ": ", custom_idx, allow_empty=bool(existing),
empty_message=" [red]Enter a number.[/red]")
if idx == 0:
return existing
if idx <= len(_REGION_PRESETS):
return _REGION_PRESETS[idx - 1][1]
custom = console.input(
" Enter your Bitwarden server URL "
"(e.g. https://vault.example.com): "
).strip()
custom = console.input(" Enter your Bitwarden server URL (e.g. https://vault.example.com): ").strip()
if not custom:
console.print(" [red]Empty URL, aborting.[/red]")
return None
if not custom.startswith(("http://", "https://")):
console.print(
" [yellow]Warning: URL doesn't start with http:// or "
"https:// — bws may reject it.[/yellow]"
)
console.print(" [yellow]Warning: URL doesn't start with http:// or https:// — bws may reject it.[/yellow]")
return custom
hermes_cli/send_cmd.py
+37 -101
View File
@@ -24,16 +24,9 @@ def _fail(msg: str, exit_code: int | None = None) -> int:
return _FAILURE_EXIT
def _read_message_body(
positional: Optional[str],
file_path: Optional[str],
) -> Optional[str]:
"""Resolve the message body from the positional arg, ``--file``, or piped stdin.
Order: explicit positional argument, then ``--file PATH`` / ``--file -`` (stdin), then piped
stdin when not attached to a TTY. Returns ``None`` when nothing is available — callers must
treat that as a usage error.
"""
def _read_message_body(positional: Optional[str], file_path: Optional[str]) -> Optional[str]:
"""Resolve the message body: positional arg, then ``--file PATH`` / ``--file -`` (stdin), then
piped stdin when not attached to a TTY. ``None`` when nothing is available (a usage error)."""
if positional:
return positional
@@ -52,33 +45,21 @@ def _read_message_body(
f' hermes send --to telegram "optional caption MEDIA:{file_path}"\n'
"Add [[as_document]] to deliver an image as an uncompressed file:\n"
f' hermes send --to telegram "[[as_document]] MEDIA:{file_path}"',
_USAGE_EXIT,
)
_USAGE_EXIT)
except OSError as exc:
_fail(f"hermes send: cannot read {file_path}: {exc}", _USAGE_EXIT)
# Piped input: only consume stdin when it is not a TTY. Reading from a
# TTY would block the user in a half-broken "type your message" state,
# which is a poor default for an ops CLI.
# Reading from a TTY would block the user in a half-broken "type your message" state.
return (sys.stdin.read() or None) if not sys.stdin.isatty() else None
def _emit_result(
result_json: str,
*,
json_mode: bool,
quiet: bool,
) -> int:
"""Print the tool result in the requested format and return the exit code.
The underlying ``send_message_tool`` always returns a JSON string. We parse it, decide
success/failure, and format accordingly.
"""
def _emit_result(result_json: str, *, json_mode: bool, quiet: bool) -> int:
"""Print the ``send_message_tool`` JSON result in the requested format; return the exit code.
Unknown / unexpected shapes are failures so scripts notice."""
try:
payload = json.loads(result_json) if result_json else {}
except json.JSONDecodeError:
# Shouldn't happen with the shared tool, but be defensive — pass the
# raw string through so the user can still see what went wrong.
# Pass the raw string through so the user can still see what went wrong.
payload = {"error": "invalid JSON from send_message_tool", "raw": result_json}
if json_mode:
@@ -89,21 +70,16 @@ def _emit_result(
elif payload.get("success"):
print(payload.get("note") or "sent")
else:
# Unknown shape — dump it so nothing is silently dropped.
print(json.dumps(payload, indent=2))
print(json.dumps(payload, indent=2)) # unknown shape — dump it, drop nothing
# Unknown / unexpected shapes are failures so scripts notice.
if not payload.get("error") and (payload.get("skipped") or payload.get("success")):
return _SUCCESS_EXIT
return _FAILURE_EXIT
def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int:
"""Print the channel directory (all configured targets across platforms).
Uses ``load_directory()`` for JSON and ``format_directory_for_display()`` for the human
rendering the send_message tool shows the model, keeping the two surfaces identical.
"""
"""Print the channel directory (all configured targets across platforms), reusing the
``format_directory_for_display`` rendering the send_message tool shows the model."""
try:
from gateway.channel_directory import format_directory_for_display, load_directory
except Exception as exc:
@@ -116,24 +92,16 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int:
platforms = dict(raw.get("platforms") or {})
# Merge in configured-but-undiscovered platforms so `--list` never hides
# a working send target. The directory only contains platforms the
# gateway has discovered channels for; a platform configured via env /
# config.yaml that has never run channel discovery (e.g. a fresh SimpleX
# setup used only for outbound `hermes send`) would otherwise be
# invisible, leaving users guessing at platform names.
# Merge in configured-but-undiscovered platforms (e.g. a fresh SimpleX setup used only for
# outbound sends) so `--list` never hides a working send target.
try:
from gateway.config import load_gateway_config
gw_config = load_gateway_config()
for plat in gw_config.get_connected_platforms():
for plat in load_gateway_config().get_connected_platforms():
plat_name = getattr(plat, "value", str(plat))
if plat_name not in ("local", "api_server", "webhook"):
platforms.setdefault(plat_name, [])
except Exception:
# Directory contents alone are still useful; don't fail --list over
# a config parse problem.
pass
pass # directory contents alone are still useful; don't fail --list on a config problem
if platform_filter:
key = platform_filter.strip().lower()
@@ -141,8 +109,7 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int:
if not filtered:
return _fail(
f"hermes send: no targets found for platform '{platform_filter}'. "
f"Configured: {', '.join(sorted(platforms)) or '(none)'}"
)
f"Configured: {', '.join(sorted(platforms)) or '(none)'}")
platforms = filtered
if json_mode:
@@ -155,9 +122,7 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int:
print("channel discovery can populate ~/.hermes/channel_directory.json.")
return _SUCCESS_EXIT
# Human display — when unfiltered, reuse the shared formatter the agent
# already sees (passing the merged view so configured-but-undiscovered
# platforms are listed too). When filtered, build a minimal view ourselves.
# Unfiltered: the shared formatter over the merged view. Filtered: a minimal view of our own.
if platform_filter is None:
print(format_directory_for_display(platforms))
return _SUCCESS_EXIT
@@ -180,10 +145,7 @@ def _list_targets(platform_filter: Optional[str], *, json_mode: bool) -> int:
def _load_hermes_env() -> None:
"""Populate ``os.environ`` from ``~/.hermes/.env`` AND bridge top-level ``config.yaml`` keys into
the environment so the underlying gateway config loader sees platform credentials and home
channel IDs.
"""
# Step 1: dotenv
the environment so the gateway config loader sees platform credentials and home channels."""
try:
from dotenv import load_dotenv
except Exception:
@@ -198,18 +160,14 @@ def _load_hermes_env() -> None:
env_path = home / ".env"
if load_dotenv and env_path.exists():
try:
# utf-8-sig strips a leading UTF-8 BOM if present (PowerShell 5.1
# Set-Content -Encoding UTF8 / Notepad) and is a no-op for
# BOM-less UTF-8. Plain "utf-8" would keep U+FEFF on the first
# key name and silently drop it from os.environ under its
# canonical name.
# utf-8-sig strips a leading BOM (PowerShell 5.1 / Notepad); plain "utf-8" would keep
# U+FEFF on the first key name and silently drop it from os.environ.
load_dotenv(str(env_path), override=True, encoding="utf-8-sig")
except UnicodeDecodeError:
try:
# utf-8-sig can't strip a BOM once we fall back to latin-1.
import codecs
import io
raw = env_path.read_bytes()
if raw.startswith(codecs.BOM_UTF8):
raw = raw[len(codecs.BOM_UTF8) :]
@@ -219,17 +177,14 @@ def _load_hermes_env() -> None:
except Exception:
pass
# Step 2: bridge top-level config.yaml values into the environment so
# gateway.config.load_gateway_config() sees them. Scalars only; don't
# override values already in the env.
# Bridge top-level config.yaml scalars into the environment (never overriding existing values).
import os
config_path = home / "config.yaml"
if not config_path.exists():
return
try:
# Presence-sensitive env bridge: raw read is deliberate — only keys
# the user actually wrote get bridged. Overlay + expansion below.
# Raw read is deliberate — only keys the user actually wrote get bridged.
from hermes_cli.config import read_user_config_raw
raw = read_user_config_raw(config_path)
except Exception:
@@ -241,8 +196,7 @@ def _load_hermes_env() -> None:
except Exception:
pass
# Managed scope: overlay administrator-pinned values before bridging to env,
# so a managed top-level scalar wins here too. Fail-open via the helper.
# Managed scope: administrator-pinned values win here too (fail-open via the helper).
try:
from hermes_cli import managed_scope
raw = managed_scope.apply_managed_overlay(raw if isinstance(raw, dict) else {})
@@ -259,18 +213,11 @@ def _load_hermes_env() -> None:
def cmd_send(args: argparse.Namespace) -> None:
"""Entry point wired into the top-level argparse dispatcher."""
_load_hermes_env() # the downstream gateway config loader reads credentials from os.environ
# Bridge ~/.hermes/.env and ~/.hermes/config.yaml into os.environ so the
# gateway config loader (invoked downstream by send_message_tool and by
# the channel directory) can see platform credentials and home channels.
_load_hermes_env()
# --list short-circuits everything else.
if getattr(args, "list_targets", False):
# When `--list telegram` is used, argparse stores "telegram" in the
# `message` positional (since list_targets takes no argument).
platform_filter = getattr(args, "message", None)
exit_code = _list_targets(platform_filter, json_mode=getattr(args, "json", False))
if getattr(args, "list_targets", False): # --list short-circuits everything else
# `hermes send --list telegram` lands "telegram" in the `message` positional.
exit_code = _list_targets(getattr(args, "message", None), json_mode=getattr(args, "json", False))
sys.exit(exit_code)
target = (getattr(args, "to", None) or "").strip()
@@ -281,32 +228,25 @@ def cmd_send(args: argparse.Namespace) -> None:
" hermes send --to telegram \"hello\"\n"
" hermes send --to discord:#ops --file report.md\n"
" hermes send --list # list available targets",
_USAGE_EXIT,
)
_USAGE_EXIT)
message = _read_message_body(getattr(args, "message", None), getattr(args, "file", None))
if message is None or not message.strip():
_fail(
"hermes send: no message provided. Pass text as a positional "
"argument, use --file PATH, or pipe data via stdin.",
_USAGE_EXIT,
)
_USAGE_EXIT)
# Optional: prepend a subject line. Useful for alerting scripts that
# want a consistent header without inlining it into every call.
# Optional subject line: a consistent header for alerting scripts.
subject = getattr(args, "subject", None)
if subject:
message = f"{subject}\n\n{message.lstrip()}"
# Import lazily so `hermes send --help` stays fast and does not pull in
# the full tool registry / gateway config stack.
# Lazy import keeps `hermes send --help` fast (no tool registry / gateway config stack).
from tools.send_message_tool import send_message_tool
# send_message_tool auto-loads gateway config + env and routes to the
# appropriate platform adapter (bot-token path for Telegram/Discord/Slack/
# Signal/SMS/WhatsApp; live-adapter path for plugin platforms).
#
# It expects the standard tool-call dict and returns a JSON string.
# Routes to the platform adapter (bot-token path for built-ins, live-adapter path for plugin
# platforms); takes the standard tool-call dict and returns a JSON string.
result = send_message_tool({"action": "send", "target": target, "message": message})
sys.exit(_emit_result(result, json_mode=getattr(args, "json", False), quiet=getattr(args, "quiet", False)))
@@ -320,19 +260,16 @@ _SEND_ARGUMENTS = (
"Delivery target. Format: 'platform' (home channel), "
"'platform:chat_id', 'platform:chat_id:thread_id', or "
"'platform:#channel-name'. Examples: telegram, "
"telegram:-1001234567890:17585, discord:#ops, slack:C0123ABCD, "
"signal:+15551234567."
"telegram:-1001234567890:17585, discord:#ops, slack:C0123ABCD, signal:+15551234567."
),
)),
(("message",), dict(nargs="?", default=None, help="Message text. If omitted, read from --file or stdin.")),
# Legacy / convenience positional removed — use --to for clarity.
(("-f", "--file"), dict(
metavar="PATH",
default=None,
help=(
"Read message body from PATH (text only). Use '-' to force stdin. "
"To send an image/document as an attachment, use MEDIA:<path> in "
"the message text instead."
"To send an image/document as an attachment, use MEDIA:<path> in the message text instead."
),
)),
(("-s", "--subject"), dict(metavar="LINE", default=None, help="Prepend a subject/header line before the message body.")),
@@ -369,8 +306,7 @@ def register_send_subparser(subparsers) -> argparse.ArgumentParser:
"\n"
"Exit codes: 0 ok, 1 delivery/backend error, 2 usage error."
),
formatter_class=argparse.RawDescriptionHelpFormatter,
)
formatter_class=argparse.RawDescriptionHelpFormatter)
for flags, kwargs in _SEND_ARGUMENTS:
parser.add_argument(*flags, **kwargs)
hermes_cli/setup.py
+158 -268
View File
@@ -1,10 +1,9 @@
"""Interactive setup wizard for Hermes Agent (config lives in ~/.hermes/).
Independently-runnable sections: Model & Provider, Terminal Backend, Agent Settings,
Messaging Platforms, Tools (TTS, web search, image generation, ...). Section bodies live in
sibling modules (setup_tts, setup_terminal, setup_platforms, setup_summary, setup_migration,
setup_quick) and are re-exported here; they resolve shared prompt/config helpers lazily through
this module so test patches on ``hermes_cli.setup.<name>`` keep working.
Independently-runnable sections: Model & Provider, Terminal Backend, Agent Settings, Messaging
Platforms, Tools. Section bodies live in sibling setup_* modules and are re-exported here; they
resolve shared prompt/config helpers lazily through this module so test patches on
``hermes_cli.setup.<name>`` keep working.
"""
import importlib.util
@@ -21,39 +20,46 @@ from typing import Callable
from hermes_cli.curses_ui import MenuNavigationEvent, MenuNavigationStart
from hermes_cli.nous_subscription import get_nous_subscription_features # noqa: F401 (re-export; patched by tests)
from tools.tool_backend_helpers import managed_nous_tools_enabled # noqa: F401 (re-export; patched by tests)
# Config helpers are re-exported (tests patch them on this module). display_hermes_home is
# imported lazily at call sites (stale-module safety during hermes update).
from hermes_cli.config import (
cfg_get, DEFAULT_CONFIG, get_hermes_home, get_config_path, get_env_path, load_config, save_config,
save_env_value, remove_env_value, get_env_value, ensure_hermes_home,
)
from hermes_cli.colors import Colors, color
from hermes_cli.cli_output import print_error, print_info, print_success, print_warning
from hermes_cli.secret_prompt import masked_secret_prompt
logger = logging.getLogger(__name__)
PROJECT_ROOT = Path(__file__).parent.parent.resolve()
_DOCS_BASE = "https://hermes-agent.nousresearch.com/docs"
_BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~")
# Config helpers (re-exported; tests patch them on this module). display_hermes_home is
# imported lazily at call sites (stale-module safety during hermes update).
from hermes_cli.config import ( # noqa: E402
cfg_get, DEFAULT_CONFIG, get_hermes_home, get_config_path, get_env_path, load_config, save_config,
save_env_value, remove_env_value, get_env_value, ensure_hermes_home,
)
from hermes_cli.colors import Colors, color # noqa: E402
def print_header(title: str):
"""Print a section header."""
def print_header(title: str, *, gap: bool = False):
"""Print a section header (``gap`` adds an extra blank line before it)."""
if gap:
print()
print()
print(color(f"◆ {title}", Colors.CYAN, Colors.BOLD))
from hermes_cli.cli_output import print_error, print_info, print_success, print_warning # noqa: E402
from hermes_cli.secret_prompt import masked_secret_prompt # noqa: E402
def _info(*lines: str | None) -> None:
"""print_info each line in order; ``None`` emits a bare blank ``print()``."""
for line in lines:
print() if line is None else print_info(line)
def _sub_dict(parent: dict, key: str) -> dict:
"""``parent[key]`` as a dict, replacing a missing or non-dict value with ``{}``."""
child = parent.get(key)
if not isinstance(child, dict):
child = parent[key] = {}
return child
def _current_reasoning_effort(config: dict) -> str:
agent_cfg = config.get("agent")
if isinstance(agent_cfg, dict):
@@ -62,20 +68,13 @@ def _current_reasoning_effort(config: dict) -> str:
def _set_reasoning_effort(config: dict, effort: str) -> None:
agent_cfg = config.get("agent")
if not isinstance(agent_cfg, dict):
agent_cfg = {}
config["agent"] = agent_cfg
agent_cfg["reasoning_effort"] = effort
_sub_dict(config, "agent")["reasoning_effort"] = effort
def is_interactive_stdin() -> bool:
"""Return True when stdin looks like a usable interactive TTY."""
stdin = getattr(sys, "stdin", None)
if stdin is None:
return False
try:
return bool(stdin.isatty())
return bool(sys.stdin.isatty())
except Exception:
return False
@@ -96,48 +95,33 @@ def print_noninteractive_setup_guidance(reason: str | None = None) -> None:
"Run 'hermes setup' in an interactive terminal to use the full wizard.", None)
_BRACKETED_PASTE_PATTERN = re.compile(r"\x1b\[\s*200~|\x1b\[\s*201~")
def _sanitize_pasted_input(value: str) -> str:
"""Strip terminal bracketed-paste control markers from pasted text."""
if not isinstance(value, str) or not value:
return value
return _BRACKETED_PASTE_PATTERN.sub("", value)
return _BRACKETED_PASTE_PATTERN.sub("", value) if isinstance(value, str) and value else value
def prompt(question: str, default: str = None, password: bool = False) -> str:
"""Prompt for input with optional default."""
display = f"{question} [{default}]: " if default else f"{question}: "
display = color(f"{question} [{default}]: " if default else f"{question}: ", Colors.YELLOW)
try:
if password:
value = masked_secret_prompt(color(display, Colors.YELLOW))
value = masked_secret_prompt(display)
else:
from hermes_cli.cli_output import line_input
value = line_input(color(display, Colors.YELLOW))
cleaned = _sanitize_pasted_input(value)
return cleaned.strip() or default or ""
value = line_input(display)
return _sanitize_pasted_input(value).strip() or default or ""
except (KeyboardInterrupt, EOFError):
print()
sys.exit(1)
# =============================================================================
# Setup navigation (Escape cancels, Left arrow goes back) — a ContextVar state
# machine shared with the curses menus.
# =============================================================================
# ── Setup navigation (Escape cancels, Left arrow goes back): a ContextVar state machine shared
# with the curses menus. ──
class _SetupControlFlow(BaseException):
"""Bypass provider error handlers that intentionally catch ``Exception``.
Provider setup has broad compatibility boundaries around network, plugin and credential
integrations; navigation must cross them unchanged so the outer state machine can replay
the prior prompt.
"""
"""Bypass provider error handlers that intentionally catch ``Exception`` so navigation reaches
the outer state machine unchanged and it can replay the prior prompt."""
class _SetupCancelled(_SetupControlFlow):
@@ -156,29 +140,22 @@ class _SetupNavigationState:
"""Per-invocation navigation state for the synchronous setup wizard."""
def __init__(self, *, section_index: int = -1, prompt_index: int = 0):
self.section_index = section_index
self.reset(section_index)
self.prompt_index = prompt_index
self.active_prompt_index = -1
self.resolved_choices: list[object] = []
self.replay_choices: list[object] = []
def reset(self, section_index: int = -1, replay: list | None = None) -> None:
"""Rewind per-section counters (entering a section, or leaving the wizard)."""
self.section_index = section_index
self.prompt_index = 0
self.active_prompt_index = -1
self.resolved_choices = []
self.replay_choices = copy.deepcopy(replay or [])
self.resolved_choices: list[object] = []
self.replay_choices: list[object] = copy.deepcopy(replay or [])
_SETUP_NAVIGATION: ContextVar[_SetupNavigationState | None] = ContextVar(
"hermes_setup_navigation", default=None
)
_SETUP_NAVIGATION: ContextVar[_SetupNavigationState | None] = ContextVar("hermes_setup_navigation", default=None)
def _handle_setup_menu_navigation(
event: MenuNavigationEvent, value: object = None
) -> MenuNavigationStart | None:
def _handle_setup_menu_navigation(event: MenuNavigationEvent, value: object = None) -> MenuNavigationStart | None:
"""Translate shared curses menu events into setup control flow."""
state = _SETUP_NAVIGATION.get()
if state is None:
@@ -187,25 +164,16 @@ def _handle_setup_menu_navigation(
if state.section_index < 0:
state.active_prompt_index = -1
return MenuNavigationStart()
state.active_prompt_index = state.prompt_index
idx = state.active_prompt_index = state.prompt_index
state.prompt_index += 1
allow_back = state.section_index > 0 or state.active_prompt_index > 0
if state.active_prompt_index < len(state.replay_choices):
return MenuNavigationStart(
allow_back=allow_back,
replay_value=copy.deepcopy(state.replay_choices[state.active_prompt_index]),
)
allow_back = state.section_index > 0 or idx > 0
if idx < len(state.replay_choices):
return MenuNavigationStart(allow_back=allow_back, replay_value=copy.deepcopy(state.replay_choices[idx]))
return MenuNavigationStart(allow_back=allow_back)
if event is MenuNavigationEvent.RESOLVE:
prompt_index = state.active_prompt_index
if prompt_index < 0:
return None
resolved = copy.deepcopy(value)
if prompt_index < len(state.resolved_choices):
state.resolved_choices[prompt_index] = resolved
del state.resolved_choices[prompt_index + 1 :]
else:
state.resolved_choices.append(resolved)
if prompt_index >= 0: # replace this answer and drop every later one
state.resolved_choices[prompt_index:] = [copy.deepcopy(value)]
return None
if event is MenuNavigationEvent.CANCEL:
raise _SetupCancelled()
@@ -228,15 +196,18 @@ def _setup_navigation_scope():
def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None:
"""Run setup sections with left-arrow navigation between choices.
Left arrow at a section's first choice returns to the previous section; from a later, nested
choice it replays earlier selections invisibly and reopens only the preceding prompt.
"""
"""Run setup sections with left-arrow navigation: at a section's first choice it returns to
the previous section; from a later choice it replays earlier selections invisibly and reopens
only the preceding prompt."""
state = _SETUP_NAVIGATION.get()
section_index = 0
answers_by_section: dict[int, list[object]] = {}
replay_by_section: dict[int, list[object]] = {}
def _record_answers() -> None:
if state is not None:
answers_by_section[section_index] = copy.deepcopy(state.resolved_choices)
try:
while section_index < len(steps):
label, action = steps[section_index]
@@ -245,8 +216,7 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None:
try:
action()
except _SetupGoBack as navigation:
if state is not None:
answers_by_section[section_index] = copy.deepcopy(state.resolved_choices)
_record_answers()
if navigation.prompt_index > 0:
previous_index = section_index
target_prompt = navigation.prompt_index - 1
@@ -254,8 +224,7 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None:
previous_index = max(0, section_index - 1)
target_prompt = max(0, len(answers_by_section.get(previous_index, [])) - 1)
replay_by_section[previous_index] = copy.deepcopy(
answers_by_section.get(previous_index, [])[:target_prompt]
)
answers_by_section.get(previous_index, [])[:target_prompt])
print()
if previous_index == section_index:
print_info(f"Returning to the previous choice in {label}...")
@@ -263,8 +232,7 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None:
print_info(f"Returning to {steps[previous_index][0]}...")
section_index = previous_index
continue
if state is not None:
answers_by_section[section_index] = copy.deepcopy(state.resolved_choices)
_record_answers()
section_index += 1
finally:
if state is not None:
@@ -274,17 +242,13 @@ def _run_setup_steps(steps: list[tuple[str, Callable[[], None]]]) -> None:
def run_setup_action_with_navigation(
label: str, action: Callable[[], None], *, cancelled_message: str = "Setup cancelled."
) -> None:
"""Run a setup-style menu flow with Escape and nested Left navigation.
Shared commands such as ``hermes model`` use the wizard's pickers outside ``run_setup_wizard``;
this installs the navigation context for them and reuses the prompt replay state machine.
"""
"""Run a setup-style menu flow with Escape and nested Left navigation — for commands such as
``hermes model`` that use the wizard's pickers outside ``run_setup_wizard``."""
with _setup_navigation_scope():
try:
_run_setup_steps([(label, action)])
except _SetupCancelled:
print()
print_info(cancelled_message)
_info(None, cancelled_message)
# ── Prompt primitives ──
@@ -297,51 +261,37 @@ def _curses_prompt_choice(question: str, choices: list, default: int = 0, descri
def prompt_choice(question: str, choices: list, default: int = 0, description: str | None = None) -> int:
"""Prompt for a choice from a list with arrow key navigation.
Escape cancels an active setup wizard; outside setup it keeps the default. The curses
component owns its own numbered fallback, so a cancel result must never be mistaken for a
request to open another prompt. Ctrl+C exits the wizard.
"""
"""Prompt for a choice from a list with arrow key navigation. Escape cancels an active setup
wizard; outside setup it keeps the default (the curses component owns its own numbered
fallback, so a cancel result must never open another prompt). Ctrl+C exits the wizard."""
idx = _curses_prompt_choice(question, choices, default, description=description)
if idx >= 0:
if idx == default:
_info(" Skipped (keeping current)", None)
return default
print()
return idx
return default
if idx < 0:
return default
if idx == default:
_info(" Skipped (keeping current)", None)
return default
print()
return idx
def is_noninteractive() -> bool:
"""True when no human is available to answer a prompt.
The dashboard/desktop spawn CLI actions with ``stdin=DEVNULL`` and ``HERMES_NONINTERACTIVE=1``
(see ``hermes_cli/web_server.py``); there ``input()`` raises ``EOFError`` immediately, and a
prompt that aborts on EOF kills the spawned action (desktop "restart gateway" failed this way
when the Windows service was not installed yet). Honour the flag so callers fall back to their
default.
"""
"""True when no human is available to answer a prompt: the dashboard/desktop spawn CLI actions
with ``stdin=DEVNULL`` and ``HERMES_NONINTERACTIVE=1`` (``hermes_cli/web_server.py``), where a
prompt that aborts on EOF would kill the spawned action — callers fall back to their default."""
return os.environ.get("HERMES_NONINTERACTIVE", "").strip().lower() in {"1", "true", "yes", "on"}
def prompt_yes_no(question: str, default: bool = True) -> bool:
"""Prompt for yes/no. Ctrl+C exits, empty input returns default.
Non-interactive callers (``HERMES_NONINTERACTIVE=1`` or a closed/redirected stdin) have no
one to answer, so fall back to ``default`` instead of aborting the whole process.
"""
"""Prompt for yes/no. Ctrl+C exits; empty input, ``HERMES_NONINTERACTIVE=1`` or a
closed/redirected stdin return ``default`` instead of aborting the whole process."""
if is_noninteractive():
return default
# Inside setup, route binary selections through the curses menu so ESC and left-arrow work
# consistently; every other caller keeps the traditional line prompt.
if _SETUP_NAVIGATION.get() is not None:
return _curses_prompt_choice(question, ["Yes", "No"], 0 if default else 1) == 0
default_str = "Y/n" if default else "y/N"
while True:
try:
value = input(color(f"{question} [{default_str}]: ", Colors.YELLOW)).strip().lower()
@@ -353,29 +303,18 @@ def prompt_yes_no(question: str, default: bool = True) -> bool:
# proceeds unattended instead of failing the whole command.
print()
return default
if not value:
return default
if value in {"y", "yes"}:
return True
if value in {"n", "no"}:
return False
answer = {"": default, "y": True, "yes": True, "n": False, "no": False}.get(value)
if answer is not None:
return answer
print_error("Please enter 'y' or 'n'")
def prompt_checklist(title: str, items: list, pre_selected: list = None) -> list:
"""Multi-select checklist; returns the sorted indices of selected items.
``pre_selected`` indices start checked; Space toggles, Enter on the appended "Continue →"
confirms, cancel keeps the pre-selection. Numbered fallback when curses is unavailable.
"""
if pre_selected is None:
pre_selected = []
"""Multi-select checklist; returns the sorted indices of selected items. ``pre_selected``
start checked; Space toggles, Enter confirms, cancel keeps the pre-selection."""
from hermes_cli.curses_ui import curses_checklist
chosen = curses_checklist(title, items, set(pre_selected), cancel_returns=set(pre_selected))
return sorted(chosen)
pre = set(pre_selected or [])
return sorted(curses_checklist(title, items, pre, cancel_returns=pre))
def _prompt_api_key(var: dict):
@@ -393,14 +332,17 @@ def _prompt_api_key(var: dict):
if var.get("url"):
print_info(f" Get your key at: {var['url']}")
print()
_prompt_and_save_env_var(var, " ✓ Saved", " Skipped (configure later with 'hermes setup')")
def _prompt_and_save_env_var(var: dict, saved_msg: str, skipped_msg: str) -> None:
"""Prompt for one env-var value (masked when secret); persist and confirm, or report the skip."""
value = prompt(f" {var.get('prompt', var['name'])}", password=bool(var.get("password")))
if value:
save_env_value(var["name"], value)
print_success(" ✓ Saved")
print_success(saved_msg)
else:
print_warning(" Skipped (configure later with 'hermes setup')")
print_warning(skipped_msg)
def _module_installed(name: str) -> bool:
@@ -419,20 +361,13 @@ def _print_banner(*lines: str) -> None:
print(color("└─────────────────────────────────────────────────────────┘", Colors.MAGENTA))
# Tool categories and provider config are in tools_config.py (shared
# between `hermes tools` and `hermes setup tools`).
# ── Section 1: Model & Provider Configuration ──
def setup_model_provider(config: dict, *, quick: bool = False):
"""Configure the inference provider and default model.
Delegates to the ``hermes model`` flow (provider picker, credential prompting, model pick,
persistence) so there is one code path — any provider added there is available here.
*quick* skips credential rotation, vision and TTS (first-time quick setup).
"""
"""Configure the inference provider and default model via the ``hermes model`` flow (one code
path — any provider added there is available here). *quick* is accepted for the first-time
quick setup caller; rotation, vision and TTS keep safe defaults either way."""
from hermes_cli.config import load_config, save_config
print_header("Inference Provider")
_info("Choose how to connect to your main chat model.",
@@ -449,8 +384,8 @@ def setup_model_provider(config: dict, *, quick: bool = False):
print_info("You can try again later with: hermes model")
# Re-sync from disk in place: cmd_model saved via its own load/save cycle and the wizard's
# final save_config(config) must not clobber it with stale values (#4172). Rotation, vision
# and TTS keep safe defaults (configure via `hermes auth add` / `hermes setup tts`).
# final save_config(config) must not clobber it with stale values. Rotation, vision and TTS
# keep safe defaults (configure via `hermes auth add` / `hermes setup tts`).
config.clear()
config.update(load_config())
save_config(config)
@@ -478,19 +413,23 @@ def _apply_default_agent_settings(config: dict):
" Run `hermes setup agent` later to customize.")
def _prompt_number(label: str, current, cast=int):
"""Prompt for a number; ``None`` when the answer does not parse."""
try:
return cast(prompt(label, str(current)))
except ValueError:
return None
def _prompt_int_setting(section: dict, key: str, label: str, current, accept) -> None:
"""Prompt for an int; store it under *key* only when it parses and *accept* holds."""
try:
value = int(prompt(label, str(current)))
if accept(value):
section[key] = value
except ValueError:
pass
value = _prompt_number(label, current)
if value is not None and accept(value):
section[key] = value
_TOOL_PROGRESS_HELP = (
"Tool Progress Display",
"Controls how much tool activity is shown (CLI and messaging).",
"Tool Progress Display", "Controls how much tool activity is shown (CLI and messaging).",
" off — Silent, just the final response",
" new — Show tool name only when it changes (less noise)",
" all — Show every tool call with a short preview",
@@ -499,14 +438,11 @@ _TOOL_PROGRESS_HELP = (
)
_SESSION_RESET_HELP = (
"Messaging sessions (Telegram, Discord, etc.) accumulate context over time.",
"Each message adds to the conversation history, which means growing API costs.",
"",
"Each message adds to the conversation history, which means growing API costs.", "",
"To manage this, sessions can automatically reset after a period of inactivity",
"or at a fixed time each day. When a reset happens, the agent saves important",
"things to its persistent memory first — but the conversation context is cleared.",
"",
"You can also manually reset anytime by typing /reset in chat.",
"",
"things to its persistent memory first — but the conversation context is cleared.", "",
"You can also manually reset anytime by typing /reset in chat.", "",
)
_SESSION_RESET_CHOICES = [
"Inactivity + daily reset (reset whichever comes first)",
@@ -530,21 +466,18 @@ def setup_agent_settings(config: dict):
"Higher = more complex tasks, but costs more tokens.",
f"Press Enter to keep {current_max}. Use 90 for most tasks or 150+ for open exploration.")
try:
max_iter = int(prompt("Max iterations", current_max))
if max_iter > 0:
# config.yaml only; gateway/run.py derives HERMES_MAX_ITERATIONS from agent.max_turns.
config.setdefault("agent", {})["max_turns"] = max_iter
config.pop("max_turns", None)
remove_env_value("HERMES_MAX_ITERATIONS")
print_success(f"Max iterations set to {max_iter}")
except ValueError:
max_iter = _prompt_number("Max iterations", current_max)
if max_iter is None:
print_warning("Invalid number, keeping current value")
elif max_iter > 0:
# config.yaml only; gateway/run.py derives HERMES_MAX_ITERATIONS from agent.max_turns.
config.setdefault("agent", {})["max_turns"] = max_iter
config.pop("max_turns", None)
remove_env_value("HERMES_MAX_ITERATIONS")
print_success(f"Max iterations set to {max_iter}")
# ── Tool Progress Display ──
print_info("")
for line in _TOOL_PROGRESS_HELP:
print_info(line)
_info("", *_TOOL_PROGRESS_HELP)
current_mode = cfg_get(config, "display", "tool_progress", default="all")
mode = prompt("Tool progress mode", current_mode)
@@ -561,34 +494,29 @@ def setup_agent_settings(config: dict):
"Higher threshold = compress later (use more context). Lower = compress sooner.")
config.setdefault("compression", {})["enabled"] = True
current_threshold = cfg_get(config, "compression", "threshold", default=0.50)
try:
threshold = float(prompt("Compression threshold (0.5-0.95)", str(current_threshold)))
if 0.5 <= threshold <= 0.95:
config["compression"]["threshold"] = threshold
except ValueError:
pass
threshold = _prompt_number("Compression threshold (0.5-0.95)", current_threshold, float)
if threshold is not None and 0.5 <= threshold <= 0.95:
config["compression"]["threshold"] = threshold
print_success(f"Context compression threshold set to {config['compression'].get('threshold', 0.50)}")
# ── Session Reset Policy ──
print_header("Session Reset Policy")
for line in _SESSION_RESET_HELP:
print_info(line)
_info(*_SESSION_RESET_HELP)
_prompt_session_reset(config.setdefault("session_reset", {}))
save_config(config)
current_policy = config.get("session_reset", {})
current_mode = current_policy.get("mode", "none")
current_idle = current_policy.get("idle_minutes", 1440)
current_hour = current_policy.get("at_hour", 4)
def _prompt_session_reset(reset_cfg: dict) -> None:
"""Pick the session reset mode and its idle/daily parameters in place."""
current_mode = reset_cfg.get("mode", "none")
current_idle, current_hour = reset_cfg.get("idle_minutes", 1440), reset_cfg.get("at_hour", 4)
default_reset = _SESSION_RESET_MODES.index(current_mode) if current_mode in _SESSION_RESET_MODES else 3
reset_idx = prompt_choice("Session reset mode:", _SESSION_RESET_CHOICES, default_reset)
reset_cfg = config.setdefault("session_reset", {})
mode = _SESSION_RESET_MODES[reset_idx] if 0 <= reset_idx < len(_SESSION_RESET_MODES) else None
if mode is not None:
reset_cfg["mode"] = mode
if mode is None: # keep current settings
return
reset_cfg["mode"] = mode
if mode in ("both", "idle"):
_prompt_int_setting(reset_cfg, "idle_minutes", " Inactivity timeout (minutes)", current_idle, lambda v: v > 0)
if mode in ("both", "daily"):
@@ -597,15 +525,13 @@ def setup_agent_settings(config: dict):
if mode == "none":
print_info("Sessions will never auto-reset. Context is managed only by compression.")
print_warning("Long conversations will grow in cost. Use /reset manually when needed.")
elif mode is not None:
else:
print_success({
"both": f"Sessions reset after {idle_now} min idle or daily at {hour_now}:00",
"idle": f"Sessions reset after {idle_now} min of inactivity",
"daily": f"Sessions reset daily at {hour_now}:00",
}[mode])
save_config(config)
# ── Section 5: Tool Configuration (delegates to unified tools_config.py) ──
@@ -621,16 +547,14 @@ def setup_tools(config: dict, first_install: bool = False):
_SEND_CONSENT_EXPLAINER = (
"",
"Sending uploads each daily package to the Nous telemetry",
"", "Sending uploads each daily package to the Nous telemetry",
"service. Packages carry your profile-scoped install ID, a",
"stable random UUID that identifies this profile across days",
"(it contains no personal information and is reset by deleting",
"the shared-metrics directory). Only packages whose entire",
"collection period falls inside a recorded consent window are",
"ever sent — data from before you opt in, or from any gap",
"while sending was off, stays on this machine. Sending can be",
"turned off again at any time.",
"while sending was off, stays on this machine. Sending can be", "turned off again at any time.",
)
@@ -640,15 +564,7 @@ def setup_telemetry(config: dict):
_info("Shared metrics contain only bounded counters and histograms.",
"Collection is local. Sending them to Nous is a separate opt-in.")
telemetry = config.get("telemetry")
if not isinstance(telemetry, dict):
telemetry = {}
config["telemetry"] = telemetry
shared_metrics = telemetry.get("shared_metrics")
if not isinstance(shared_metrics, dict):
shared_metrics = {}
telemetry["shared_metrics"] = shared_metrics
shared_metrics = _sub_dict(_sub_dict(config, "telemetry"), "shared_metrics")
current = shared_metrics.get("enabled") is True
shared_metrics["enabled"] = prompt_yes_no("Enable local shared metrics?", default=current)
if not shared_metrics["enabled"]:
@@ -663,8 +579,7 @@ def setup_telemetry(config: dict):
return
print_success("Local shared metrics enabled.")
for line in _SEND_CONSENT_EXPLAINER:
print_info(line)
_info(*_SEND_CONSENT_EXPLAINER)
shared_metrics["send"] = prompt_yes_no("Send shared metrics to Nous?", default=shared_metrics.get("send") is True)
_record_send_consent_change(enabled=shared_metrics["send"])
if shared_metrics["send"]:
@@ -674,20 +589,14 @@ def setup_telemetry(config: dict):
def _record_send_consent_change(*, enabled: bool) -> None:
"""Reconcile consent windows at the moment the user decides.
Same single writer as the relay and the sender, so wizard, relay and mid-pass callers cannot
disagree; the relay would reconcile on its next hook anyway — this makes the effect immediate.
"""
"""Reconcile consent windows at the moment the user decides — same single writer as the relay
and the sender, so wizard, relay and mid-pass callers cannot disagree."""
try:
from hermes_cli.observability.shared_metrics import SharedMetricsStore
from hermes_cli.observability.shared_metrics_sender import reconcile_send_consent
from hermes_cli.sqlite_util import write_txn
store = SharedMetricsStore()
with store._connection() as connection:
with write_txn(connection):
reconcile_send_consent(connection, enabled)
with SharedMetricsStore()._connection() as connection, write_txn(connection):
reconcile_send_consent(connection, enabled)
except Exception:
# Never block the wizard on telemetry bookkeeping; the relay reconciles on the next hook.
logger.debug("Unable to record shared-metrics consent change", exc_info=True)
@@ -743,7 +652,7 @@ def run_setup_wizard(args):
def _backup_config_file(config_path: Path) -> Path | None:
"""Back up config.yaml before setup modifies it (#3522); None when absent or copy fails."""
"""Back up config.yaml before setup modifies it; None when absent or copy fails."""
if not config_path.exists():
return None
from datetime import datetime as _dt
@@ -790,46 +699,32 @@ def _run_full_setup(config: dict, hermes_home, *, is_existing: bool, migration_r
def _skip(key: str, label: str) -> bool:
return migration_ran and _skip_configured_section(config, key, label)
def _model_step() -> None:
if not _skip("model", "Model & Provider"):
setup_model_provider(config)
def _terminal_step() -> None:
if not _skip("terminal", "Terminal Backend"):
setup_terminal_backend(config)
def _gateway_step() -> None:
if not _skip("gateway", "Messaging Platforms"):
setup_gateway(config)
return
# A skipped (migrated) gateway section still needs its service so imported platforms
# and cron jobs become active.
from hermes_cli.gateway import ensure_gateway_service
ensure_gateway_service(context="setup")
def _tools_step() -> None:
if not _skip("tools", "Tools"):
setup_tools(config, first_install=not is_existing)
def _step(key: str, label: str, run) -> tuple:
return label, lambda: None if _skip(key, label) else run()
_run_setup_steps(
[
("Model & Provider", _model_step),
("Terminal Backend", _terminal_step),
("Messaging Platforms", _gateway_step),
("Tools", _tools_step),
]
)
_run_setup_steps([
_step("model", "Model & Provider", lambda: setup_model_provider(config)),
_step("terminal", "Terminal Backend", lambda: setup_terminal_backend(config)),
("Messaging Platforms", _gateway_step),
_step("tools", "Tools", lambda: setup_tools(config, first_install=not is_existing))])
# First-time mode picker: (menu label, runner) — a None runner falls through to Full Setup.
_FIRST_TIME_MODES = (
("Quick Setup (Nous Portal) — free OAuth login, no API keys, model + tools (recommended)",
lambda config, home, is_existing: _run_first_time_quick_setup(config, home, is_existing)),
lambda *a: _run_first_time_quick_setup(*a)),
("Full setup — configure every provider, tool & option yourself (bring your own keys)", None),
("Blank Slate — everything off except the bare minimum; opt in to each capability",
lambda config, home, is_existing: _run_blank_slate_setup(config, home, is_existing)),
lambda *a: _run_blank_slate_setup(*a)),
)
@@ -867,16 +762,12 @@ def _run_setup_wizard_impl(args):
# Existing installation == a provider is configured
from hermes_cli.auth import get_active_provider
is_existing = bool(
get_env_value("OPENROUTER_API_KEY") or get_env_value("OPENAI_BASE_URL") or get_active_provider() is not None
)
_print_banner(
"│ ⚕ Hermes Agent Setup Wizard │",
"├─────────────────────────────────────────────────────────┤",
"│ Let's configure your Hermes Agent installation. │",
"│ Press Ctrl+C at any time to exit. │",
)
is_existing = bool(get_env_value("OPENROUTER_API_KEY") or get_env_value("OPENAI_BASE_URL")
or get_active_provider() is not None)
_print_banner("│ ⚕ Hermes Agent Setup Wizard │",
"├─────────────────────────────────────────────────────────┤",
"│ Let's configure your Hermes Agent installation. │",
"│ Press Ctrl+C at any time to exit. │")
migration_ran = False
if is_existing:
@@ -886,8 +777,7 @@ def _run_setup_wizard_impl(args):
if quick_requested:
_run_setup_steps([("Quick Setup", lambda: _run_quick_setup(config, hermes_home))])
return
print()
print_header("Reconfigure")
print_header("Reconfigure", gap=True)
print_success("You already have Hermes configured.")
_info("Running the full wizard — each prompt shows your current value.",
"Press Enter to keep it, or type a new value to change it.", "",
hermes_cli/setup_hidden_env.py
+12 -18
View File
@@ -1,25 +1,21 @@
"""Which platform env vars the setup surfaces hide.
Hiding them is a *presentation* decision only. The env vars keep working through ``hermes config
set``, ``.env``, and ``config.yaml``; the gateway reads them exactly as before. This module just
says what a new user is asked during setup.
Hiding is a *presentation* decision only: the vars keep working through ``hermes config set``,
``.env`` and ``config.yaml``, and the gateway reads them exactly as before.
"""
# Suffix match, so plugin adapters nobody enumerated (IRC, SimpleX, LINE, ntfy)
# get the same treatment without a code change here.
# Suffix match, so plugin adapters nobody enumerated (IRC, SimpleX, LINE, ntfy) get the same
# treatment without a code change here.
#
# *_HOME_CHANNEL* the bot offers /sethome on the first chat
# *_ALLOW_ALL_USERS defaults off; enabling it is a security decision
# *_REPLY_TO_MODE cosmetic threading preference
# *_REPLY_MODE same, Mattermost's spelling
# *_REQUIRE_MENTION behavior toggle with a sane default
# *_AUTO_THREAD same
# *_FREE_RESPONSE_* per-channel tuning, done once the bot is in a server
# *_ALLOWED_CHANNELS same
# *_REPLY_TO_MODE / *_REPLY_MODE cosmetic threading preference (Mattermost spelling too)
# *_REQUIRE_MENTION / *_AUTO_THREAD behavior toggles with sane defaults
# *_FREE_RESPONSE_* / *_ALLOWED_CHANNELS per-channel tuning, done once the bot is in a server
# *_PROXY only for networks that block the platform
#
# Allowlists (*_ALLOWED_USERS) deliberately stay visible: that IS the decision
# a new user has to make, and the gateway denies everyone until it's set.
# Allowlists (*_ALLOWED_USERS) deliberately stay visible: that IS the decision a new user has to
# make, and the gateway denies everyone until it's set.
SETUP_HIDDEN_ENV_SUFFIXES = (
"_HOME_CHANNEL",
"_HOME_CHANNEL_NAME",
@@ -38,9 +34,7 @@ SETUP_HIDDEN_ENV_SUFFIXES = (
def is_setup_hidden_env(name: str) -> bool:
"""True when a var is self-configuring and shouldn't appear in setup forms.
Callers must still keep any var a platform lists as *required* — hiding a required credential
would make that platform unconfigurable from the UI.
"""
"""True when a var is self-configuring and shouldn't appear in setup forms. Callers must still
keep any var a platform lists as *required* — hiding a required credential would make that
platform unconfigurable from the UI."""
return name.endswith(SETUP_HIDDEN_ENV_SUFFIXES)
hermes_cli/setup_migration.py
+56 -64
View File
@@ -1,7 +1,5 @@
"""Post-migration section-skip logic and the OpenClaw first-run migration flow.
Extracted from hermes_cli/setup.py, which re-exports the names it still uses.
"""
"""Post-migration section-skip logic and the OpenClaw first-run migration flow (setup.py
re-exports the names it still uses; setup helpers are imported lazily so test patches apply)."""
import importlib.util
import logging
@@ -33,7 +31,6 @@ def _model_section_has_credentials(config: dict) -> bool:
return True
except Exception:
pass
try:
from hermes_cli.auth import PROVIDER_REGISTRY
except Exception:
@@ -44,8 +41,7 @@ def _model_section_has_credentials(config: dict) -> bool:
# mirrors is_provider_explicitly_configured in auth.py.
return any(get_env_value(v) for v in pconfig.api_key_env_vars if v != "CLAUDE_CODE_OAUTH_TOKEN")
def _any_openrouter_key() -> bool:
return any(get_env_value(v) for v in _OPENROUTER_ENV_VARS)
any_openrouter_key = any(get_env_value(v) for v in _OPENROUTER_ENV_VARS)
# Prefer the provider declared in config.yaml, avoids false positives from stray
# env vars (GH_TOKEN, etc.) when the user has already picked a different provider.
@@ -54,23 +50,17 @@ def _model_section_has_credentials(config: dict) -> bool:
provider_id = (model_cfg.get("provider") or "").strip().lower()
if provider_id in PROVIDER_REGISTRY and _has_key(PROVIDER_REGISTRY[provider_id]):
return True
if provider_id == "openrouter" and _any_openrouter_key():
if provider_id == "openrouter" and any_openrouter_key:
return True
# OpenRouter aggregator fallback (no provider declared in config).
if _any_openrouter_key():
if any_openrouter_key:
return True
# Skip copilot in auto-detect: GH_TOKEN / GITHUB_TOKEN are commonly set for
# git tooling. Mirrors resolve_provider in auth.py.
# Skip copilot in auto-detect: GH_TOKEN / GITHUB_TOKEN are commonly set for git tooling.
# Mirrors resolve_provider in auth.py.
return any(_has_key(pconfig) for pid, pconfig in PROVIDER_REGISTRY.items() if pid != "copilot")
def _gateway_platform_short_label(label: str) -> str:
"""Strip trailing parenthetical qualifiers from a gateway platform label."""
return label.split("(", 1)[0].strip() or label
def _model_summary(config: dict) -> Optional[str]:
if not _model_section_has_credentials(config):
return None
@@ -92,7 +82,8 @@ def _gateway_summary(config: dict) -> Optional[str]:
# Any non-empty status other than "not configured" counts — WhatsApp ("enabled, not paired"),
# Matrix ("configured + E2EE"), Signal ("partially configured") mean setup already started.
configured = [
_gateway_platform_short_label(plat["label"])
# Trailing parenthetical qualifiers are stripped from the label.
plat["label"].split("(", 1)[0].strip() or plat["label"]
for plat in _all_platforms()
if _platform_status(plat) and _platform_status(plat) != "not configured"
]
@@ -122,20 +113,15 @@ _SECTION_SUMMARIES = {
def _get_section_config_summary(config: dict, section_key: str) -> Optional[str]:
"""Return a short summary if a setup section is already configured, else None.
Used after OpenClaw migration to detect which sections can be skipped. ``get_env_value`` is
reached through hermes_cli.setup so that test patches on ``setup_mod.get_env_value`` apply.
"""
"""Short summary if a setup section is already configured (post-OpenClaw-migration skip
detection), else None. ``get_env_value`` is reached through hermes_cli.setup so test patches
on ``setup_mod.get_env_value`` apply."""
summarize = _SECTION_SUMMARIES.get(section_key)
return summarize(config) if summarize else None
def _skip_configured_section(config: dict, section_key: str, label: str) -> bool:
"""Show an already-configured section summary and offer to skip.
Returns True if the user chose to skip, False if the section should run.
"""
"""Show an already-configured section summary and offer to skip; True when the user skips."""
from hermes_cli.setup import print_success, prompt_yes_no
summary = _get_section_config_summary(config, section_key)
if not summary:
@@ -162,8 +148,7 @@ def _load_openclaw_migration_module():
if spec is None or spec.loader is None:
return None
mod = importlib.util.module_from_spec(spec)
# Register in sys.modules so @dataclass can resolve the module
# (Python 3.11+ requires this for dynamically loaded modules)
# Registered in sys.modules so @dataclass can resolve the module (Python 3.11+ requirement).
sys.modules[spec.name] = mod
try:
spec.loader.exec_module(mod)
@@ -232,8 +217,7 @@ def _print_migration_preview(report: dict):
if status == "migrated": # collect warnings for high-impact items
kind_lower, dest_lower = kind.lower(), str(item.get("destination", "")).lower()
warnings_shown.update(
w for kw, w in _HIGH_IMPACT_KIND_KEYWORDS.items() if kw in kind_lower or kw in dest_lower
)
w for kw, w in _HIGH_IMPACT_KIND_KEYWORDS.items() if kw in kind_lower or kw in dest_lower)
print()
if warnings_shown:
@@ -255,6 +239,21 @@ def _run_migrator(mod, openclaw_dir: Path, hermes_home: Path, selected, *, execu
).migrate()
_FAILED = object()
def _migration_step(label: str, log_label: str, fn):
"""Run one migration phase; on failure warn ``"<label>: <exc>"``, log the trace and return
``_FAILED`` (the caller aborts)."""
from hermes_cli.setup import print_warning
try:
return fn()
except Exception as e:
print_warning(f"{label}: {e}")
logger.debug(log_label, exc_info=True)
return _FAILED
def _offer_openclaw_migration(hermes_home: Path) -> bool:
"""Detect ~/.openclaw and offer to migrate during first-time setup: dry-run preview first,
execute only after explicit confirmation. Returns True iff migration ran successfully."""
@@ -266,8 +265,7 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool:
if not openclaw_dir.is_dir() or not _OPENCLAW_SCRIPT.exists():
return False
print()
print_header("OpenClaw Installation Detected")
print_header("OpenClaw Installation Detected", gap=True)
_info(f"Found OpenClaw data at {openclaw_dir}",
"Hermes can preview what would be imported before making any changes.", None)
if not prompt_yes_no("Would you like to see what can be imported?", default=True):
@@ -278,31 +276,28 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool:
if not get_config_path().exists():
save_config(load_config())
try:
mod = _load_openclaw_migration_module()
if mod is None:
print_warning("Could not load migration script.")
return False
except Exception as e:
print_warning(f"Could not load migration script: {e}")
logger.debug("OpenClaw migration module load error", exc_info=True)
mod = _migration_step("Could not load migration script", "OpenClaw migration module load error",
_load_openclaw_migration_module)
if mod is None:
print_warning("Could not load migration script.")
if mod is None or mod is _FAILED:
return False
# ── Phase 1: Dry-run preview (overwrite=True shows everything, including conflicts) ──
try:
def _preview():
selected = mod.resolve_selected_options(None, None, preset="full")
preview_report = _run_migrator(mod, openclaw_dir, hermes_home, selected, execute=False, overwrite=True)
except Exception as e:
print_warning(f"Migration preview failed: {e}")
logger.debug("OpenClaw migration preview error", exc_info=True)
return selected, _run_migrator(mod, openclaw_dir, hermes_home, selected, execute=False, overwrite=True)
previewed = _migration_step("Migration preview failed", "OpenClaw migration preview error", _preview)
if previewed is _FAILED:
return False
selected, preview_report = previewed
preview_count = preview_report.get("summary", {}).get("migrated", 0)
if preview_count == 0:
_info(None, "Nothing to import from OpenClaw.")
return False
print()
print_header(f"Migration Preview — {preview_count} item(s) would be imported")
print_header(f"Migration Preview — {preview_count} item(s) would be imported", gap=True)
_info("No changes have been made yet. Review the list below:", None)
_print_migration_preview(preview_report)
@@ -312,28 +307,25 @@ def _offer_openclaw_migration(hermes_home: Path) -> bool:
"Use --dry-run to preview again, or --preset minimal for a lighter import.")
return False
# overwrite=False so existing Hermes configs are preserved. The user saw the
# preview; conflicts are skipped by default.
try:
report = _run_migrator(mod, openclaw_dir, hermes_home, selected, execute=True, overwrite=False)
except Exception as e:
print_warning(f"Migration failed: {e}")
logger.debug("OpenClaw migration error", exc_info=True)
# overwrite=False so existing Hermes configs are preserved. The user saw the preview;
# conflicts are skipped by default.
report = _migration_step("Migration failed", "OpenClaw migration error", lambda: _run_migrator(
mod, openclaw_dir, hermes_home, selected, execute=True, overwrite=False))
if report is _FAILED:
return False
summary = report.get("summary", {})
migrated, skipped = summary.get("migrated", 0), summary.get("skipped", 0)
conflicts, errors = summary.get("conflict", 0), summary.get("error", 0)
print()
for count, printer, text in (
(migrated, print_success, f"Imported {migrated} item(s) from OpenClaw."),
(conflicts, print_info,
f"Skipped {conflicts} item(s) that already exist in Hermes (use hermes claw migrate --overwrite to force)."),
(skipped, print_info, f"Skipped {skipped} item(s) (not found or unchanged)."),
(errors, print_warning, f"{errors} item(s) had errors — check the migration report."),
for key, printer, text in (
("migrated", print_success, "Imported {n} item(s) from OpenClaw."),
("conflict", print_info,
"Skipped {n} item(s) that already exist in Hermes (use hermes claw migrate --overwrite to force)."),
("skipped", print_info, "Skipped {n} item(s) (not found or unchanged)."),
("error", print_warning, "{n} item(s) had errors — check the migration report."),
):
count = summary.get(key, 0)
if count:
printer(text)
printer(text.format(n=count))
output_dir = report.get("output_dir")
if output_dir:
hermes_cli/setup_platforms.py
+44 -55
View File
@@ -1,6 +1,6 @@
"""Messaging-platform setup wizards (Telegram, BlueBubbles, QQ Bot, webhooks) and the
``hermes setup gateway`` flow. Extracted from hermes_cli/setup.py; setup.py re-exports
the public names, and tests monkeypatch prompt/print/env helpers on hermes_cli.setup."""
"""Messaging-platform setup wizards (Telegram, BlueBubbles, webhooks) and the ``hermes setup
gateway`` flow. setup.py re-exports the public names, and tests monkeypatch prompt/print/env
helpers on hermes_cli.setup, so those are imported lazily per function."""
import logging
import re
@@ -9,6 +9,7 @@ from pathlib import Path
logger = logging.getLogger("hermes_cli.setup")
_TELEGRAM_BOT_TOKEN_RE = re.compile(r"^\d+:[A-Za-z0-9_-]{30,}$")
_RULE = "━" * 50
def _is_valid_telegram_bot_token(token: str) -> bool:
@@ -17,9 +18,7 @@ def _is_valid_telegram_bot_token(token: str) -> bool:
def _profile_name_from_hermes_home(hermes_home) -> str | None:
"""Return the active profile name when HERMES_HOME is a profile dir."""
if hermes_home.parent.name == "profiles":
return hermes_home.name
return None
return hermes_home.name if hermes_home.parent.name == "profiles" else None
def _setup_telegram_auto_result():
@@ -47,10 +46,21 @@ def _declines_reconfigure(env_var: str, label: str, question: str) -> bool:
return not prompt_yes_no(question, False)
def _save_if_set(env_var: str, value: str) -> None:
from hermes_cli.setup import save_env_value
def _save_prompted(env_var: str, question: str, *, password: bool = False, success_msg: str | None = None,
skip_msg: str | None = None, transform=None) -> str:
"""Prompt, persist the (optionally transformed) answer when non-empty, and report either way.
``success_msg`` may reference ``{value}``. Returns the raw answer ("" when skipped).
"""
from hermes_cli.setup import print_success, print_warning, prompt, save_env_value
value = prompt(question, password=password)
if value:
save_env_value(env_var, value)
save_env_value(env_var, transform(value) if transform else value)
if success_msg:
print_success(success_msg.format(value=value))
elif skip_msg:
print_warning(skip_msg)
return value
def _save_allowlist(env_var: str, users: str, success_msg: str) -> None:
@@ -77,13 +87,10 @@ def _prompt_telegram_bot_token() -> str | None:
print_info("Create a bot via @BotFather on Telegram")
while True:
token = prompt("Telegram bot token", password=True)
if not token:
return None
if not _is_valid_telegram_bot_token(token):
print_error("Invalid token format. Expected: <numeric_id>:<alphanumeric_hash> "
"(e.g., 123456789:ABCdefGHI-jklMNOpqrSTUvwxYZ)")
continue
return token
if not token or _is_valid_telegram_bot_token(token):
return token or None
print_error("Invalid token format. Expected: <numeric_id>:<alphanumeric_hash> "
"(e.g., 123456789:ABCdefGHI-jklMNOpqrSTUvwxYZ)")
def _telegram_allowlist_nudge() -> None:
@@ -109,18 +116,14 @@ def _obtain_telegram_token():
" [2] Manual",
" Create a bot via @BotFather yourself and paste the token.", None)
choice = prompt("Choice [1/2]", default="1")
token = None
setup_result = None
if choice.strip() == "1":
token = setup_result = None
if prompt("Choice [1/2]", default="1").strip() == "1":
setup_result = _setup_telegram_auto_result()
if setup_result:
token = setup_result.token
if not _is_valid_telegram_bot_token(token):
print_error("Automatic setup returned an invalid Telegram bot token.")
token = None
setup_result = None
token = setup_result = None
if not token:
_info(None, "Falling back to manual setup...", None)
@@ -131,8 +134,7 @@ def _obtain_telegram_token():
def _setup_telegram():
"""Configure Telegram bot credentials and allowlist."""
from hermes_cli.setup import print_header, print_info, print_success, prompt, prompt_yes_no
from hermes_cli.setup import _info, save_env_value
from hermes_cli.setup import _info, print_info, print_header, print_success, prompt, prompt_yes_no, save_env_value
print_header("Telegram")
if _declines_reconfigure("TELEGRAM_BOT_TOKEN", "Telegram", "Reconfigure Telegram?"):
_telegram_allowlist_nudge()
@@ -156,11 +158,7 @@ def _setup_telegram():
print_success(f"Detected your Telegram user ID: {detected_id}")
if prompt_yes_no("Allow this Telegram account to use the bot?", True):
extra = prompt("Additional allowed user IDs (comma-separated, optional)")
ids = [detected_id]
for uid in extra.replace(" ", "").split(","):
if uid and uid not in ids:
ids.append(uid)
allowed_users = ",".join(ids)
allowed_users = ",".join(dict.fromkeys([detected_id, *filter(None, extra.replace(" ", "").split(","))]))
if allowed_users is None:
allowed_users = prompt("Allowed user IDs (comma-separated, leave empty for open access)")
@@ -177,19 +175,17 @@ def _setup_telegram():
first_user_id = allowed_users.split(",")[0].strip() if allowed_users else ""
if not first_user_id:
print_info(" You can also set this later by typing /set-home in your Telegram chat.")
_save_if_set("TELEGRAM_HOME_CHANNEL", prompt("Home channel ID (leave empty to set later)"))
_save_prompted("TELEGRAM_HOME_CHANNEL", "Home channel ID (leave empty to set later)")
elif prompt_yes_no(f"Use your user ID ({first_user_id}) as the home channel?", True):
save_env_value("TELEGRAM_HOME_CHANNEL", first_user_id)
print_success(f"Telegram home channel set to {first_user_id}")
else:
q = "Home channel ID (or leave empty to set later with /set-home in Telegram)"
_save_if_set("TELEGRAM_HOME_CHANNEL", prompt(q))
_save_prompted("TELEGRAM_HOME_CHANNEL", "Home channel ID (or leave empty to set later with /set-home in Telegram)")
def _setup_bluebubbles():
"""Configure BlueBubbles iMessage gateway."""
from hermes_cli.setup import print_header, print_info, print_success, print_warning, prompt
from hermes_cli.setup import _info, prompt_yes_no, save_env_value
from hermes_cli.setup import _info, print_header, print_info, print_success, prompt, prompt_yes_no
print_header("BlueBubbles (iMessage)")
if _declines_reconfigure("BLUEBUBBLES_SERVER_URL", "BlueBubbles", "Reconfigure BlueBubbles?"):
return
@@ -200,15 +196,14 @@ def _setup_bluebubbles():
" Download: https://bluebubbles.app/", None,
"In BlueBubbles Server → Settings → API, note your Server URL and Password.", None)
for label, env_var, secret, what in (
("BlueBubbles server URL (e.g. http://192.168.1.10:1234)", "BLUEBUBBLES_SERVER_URL", False, "Server URL"),
("BlueBubbles server password", "BLUEBUBBLES_PASSWORD", True, "Password"),
for label, env_var, secret, what, transform in (
("BlueBubbles server URL (e.g. http://192.168.1.10:1234)", "BLUEBUBBLES_SERVER_URL", False, "Server URL",
lambda v: v.rstrip("/")),
("BlueBubbles server password", "BLUEBUBBLES_PASSWORD", True, "Password", None),
):
value = prompt(label, password=secret)
if not value:
print_warning(f"{what} is required — skipping BlueBubbles setup")
if not _save_prompted(env_var, label, password=secret, transform=transform,
skip_msg=f"{what} is required — skipping BlueBubbles setup"):
return
save_env_value(env_var, value.rstrip("/") if env_var == "BLUEBUBBLES_SERVER_URL" else value)
print_success("BlueBubbles credentials saved")
_info(None, "🔒 Security: Restrict who can message your bot",
@@ -221,7 +216,7 @@ def _setup_bluebubbles():
_info(None, "📬 Home Channel: phone or email for cron job delivery and notifications.",
" You can also set this later with /set-home in your iMessage chat.")
_save_if_set("BLUEBUBBLES_HOME_CHANNEL", prompt("Home channel address (leave empty to set later)"))
_save_prompted("BLUEBUBBLES_HOME_CHANNEL", "Home channel address (leave empty to set later)")
_info(None, "Advanced settings (defaults are fine for most setups):")
if prompt_yes_no("Configure webhook listener settings?", False):
@@ -247,13 +242,9 @@ def _setup_webhooks():
_info(" Full guide: https://hermes-agent.nousresearch.com/docs/user-guide/messaging/webhooks/", None)
_save_port("WEBHOOK_PORT", prompt("Webhook port (default 8644)"), "8644")
secret = prompt("Global HMAC secret (shared across all routes)", password=True)
if secret:
save_env_value("WEBHOOK_SECRET", secret)
print_success("Webhook secret saved")
else:
print_warning("No secret set — you must configure per-route secrets in config.yaml")
_save_prompted("WEBHOOK_SECRET", "Global HMAC secret (shared across all routes)", password=True,
success_msg="Webhook secret saved",
skip_msg="No secret set — you must configure per-route secrets in config.yaml")
save_env_value("WEBHOOK_ENABLED", "true")
print()
@@ -291,8 +282,7 @@ def _warn_missing_home_channels() -> None:
from hermes_cli.setup import get_env_value, _info, print_warning
missing_home = [
plat for plat, token_var, home_vars in _HOME_CHANNEL_CHECKS
if get_env_value(token_var) and not any(get_env_value(v) for v in home_vars)
]
if get_env_value(token_var) and not any(get_env_value(v) for v in home_vars)]
if not missing_home:
return
print()
@@ -364,7 +354,7 @@ def setup_gateway(config: dict):
any_messaging = any(_is_progress(_platform_status(p)) for p in _all_platforms())
if any_messaging:
print()
print_info("━" * 50)
print_info(_RULE)
print_success("Messaging platforms configured!")
_warn_missing_home_channels()
@@ -373,7 +363,6 @@ def setup_gateway(config: dict):
# `hermes setup gateway`). Gating it on messaging config left install-then-import machines
# with cron jobs and bot tokens but no process to serve them.
from hermes_cli.gateway import _is_service_running, supports_systemd_services, ensure_gateway_service
supports_systemd = supports_systemd_services()
print()
@@ -383,4 +372,4 @@ def setup_gateway(config: dict):
# Not running: install (if needed) and start, no questions asked.
ensure_gateway_service(context="setup")
print_info("━" * 50)
print_info(_RULE)
hermes_cli/setup_quick.py
+131 -172
View File
@@ -1,7 +1,6 @@
"""Streamlined setup flows extracted from hermes_cli/setup.py: the Nous Portal
one-shot (`hermes portal`), first-time quick setup, Blank Slate setup and the
`--quick` missing-items pass. Names from setup.py are imported lazily per
function so test patches on ``hermes_cli.setup`` take effect."""
"""Streamlined setup flows: the Nous Portal one-shot (`hermes portal`), first-time quick setup,
Blank Slate setup and the `--quick` missing-items pass. Names from setup.py are imported lazily
per function so test patches on ``hermes_cli.setup`` take effect."""
import logging
import os
@@ -13,65 +12,62 @@ logger = logging.getLogger("hermes_cli.setup")
# (env-var name substring, platform label, emoji) — order matters: first match wins.
_MESSAGING_PLATFORMS = (("TELEGRAM", "Telegram", "📱"), ("DISCORD", "Discord", "💬"), ("SLACK", "Slack", "💼"))
_BLANK_SLATE_DONE_LINES = (
" Seed skills: hermes skills opt-in --sync", " Add MCP servers: hermes mcp add",
)
def _reload_config_into(config: dict) -> None:
"""Re-sync the in-memory config dict from disk after a sub-flow that saved
via its own load/save cycle, so a later save_config(config) can't clobber it."""
def _reload_config_into(config: dict, *, dict_only: bool = False) -> None:
"""Re-sync the in-memory config dict from disk after a sub-flow that saved via its own
load/save cycle, so a later save_config(config) can't clobber it."""
from hermes_cli.setup import load_config
_refreshed = load_config()
config.clear()
config.update(_refreshed)
refreshed = load_config()
if not dict_only or isinstance(refreshed, dict):
config.clear()
config.update(refreshed)
def _prompt_and_save_env_var(var: dict, saved_msg: str, skipped_msg: str) -> None:
"""Prompt for one env-var value (masked when secret); persist and confirm, or report the skip."""
from hermes_cli.setup import print_success, print_warning, prompt, save_env_value
value = prompt(f" {var.get('prompt', var['name'])}", password=bool(var.get("password")))
if value:
save_env_value(var["name"], value)
print_success(saved_msg)
else:
print_warning(skipped_msg)
def _run_nous_flow(config: dict, *, context: str, cancel_exc: tuple, cancel_lines: tuple, print_error) -> bool:
"""Run ``_model_flow_nous`` (login, model pick, provider switch, Tool Gateway opt-in) — the
single source of truth shared with ``hermes model``. False when cancelled or failed (the
message is already printed)."""
from hermes_cli.setup import _info
try:
from hermes_cli.main import _model_flow_nous
_model_flow_nous(config)
return True
except cancel_exc:
# _login_nous raises SystemExit(130)/(1) on cancel/failure; the expired-session re-login
# path inside _model_flow_nous only catches Exception, so SystemExit would kill the CLI.
_info(*cancel_lines)
except Exception as exc:
logger.debug("_model_flow_nous error during %s: %s", context, exc)
print_error(exc)
return False
def _run_portal_one_shot(config: dict) -> None:
"""One-shot Nous Portal setup (``hermes setup --portal`` / ``hermes portal``).
Login, model pick, provider switch and Tool Gateway opt-in are all delegated to
``_model_flow_nous`` — the same flow quick setup and ``hermes model`` use for Nous — so
there is one source of truth and ``hermes portal`` always offers a picker.
"""
from hermes_cli.setup import color, Colors, _info, load_config, print_error, print_info, print_success
print()
print(color("┌─────────────────────────────────────────────────────────┐", Colors.MAGENTA))
print(color("│ ⚕ Hermes Setup — Nous Portal (one-shot) │", Colors.MAGENTA))
print(color("└─────────────────────────────────────────────────────────┘", Colors.MAGENTA))
"""One-shot Nous Portal setup (``hermes setup --portal`` / ``hermes portal``)."""
from hermes_cli.setup import _info, _print_banner, print_error, print_info, print_success
_print_banner("│ ⚕ Hermes Setup — Nous Portal (one-shot) │")
_info(None, " One subscription, 300+ models, plus the Tool Gateway:",
" web search, image generation, TTS, browser automation",
" — all routed through your Nous Portal sub.", None,
" Sign up: https://portal.nousresearch.com/manage-subscription", None)
try:
from hermes_cli.main import _model_flow_nous
_model_flow_nous(config)
except (KeyboardInterrupt, EOFError, SystemExit):
# _login_nous raises SystemExit(130)/(1) on cancel/failure; the expired-session re-login
# path inside _model_flow_nous only catches Exception, so SystemExit would kill the CLI.
_info(None, " Setup cancelled.", " You can retry later with `hermes portal`.")
return
except Exception as exc:
logger.debug("_model_flow_nous error during `hermes portal`: %s", exc)
def _on_error(exc: Exception) -> None:
print()
print_error(f" Nous Portal setup encountered an error: {exc}")
print_info(" You can retry later with `hermes portal`.")
if not _run_nous_flow(config, context="`hermes portal`", cancel_exc=(KeyboardInterrupt, EOFError, SystemExit),
cancel_lines=(None, " Setup cancelled.", " You can retry later with `hermes portal`."),
print_error=_on_error):
return
# Re-sync from disk so a caller's later save_config(config) can't clobber the login save.
try:
_refreshed = load_config()
if isinstance(_refreshed, dict):
config.clear()
config.update(_refreshed)
_reload_config_into(config, dict_only=True)
except Exception:
pass
@@ -81,33 +77,26 @@ def _run_portal_one_shot(config: dict) -> None:
def _run_first_time_quick_setup(config: dict, hermes_home, is_existing: bool):
"""Streamlined first-time setup via Nous Portal: OAuth, model, terminal & messaging.
Everything else gets sensible defaults; customize later via ``hermes setup
<section>`` or switch providers with ``hermes model``.
"""
"""Streamlined first-time setup via Nous Portal: OAuth, model, terminal & messaging;
everything else gets defaults."""
from hermes_cli.setup import (
_apply_default_agent_settings, _info, print_header, print_info, _print_macos_fda_tip,
_print_setup_summary, print_success, print_warning, prompt_choice, save_config, setup_gateway,
setup_terminal_backend,
)
# Step 1: Nous Portal — OAuth login + model selection (provider set to "nous" by the save).
print()
print_header("Nous Portal")
print_header("Nous Portal", gap=True)
_info("One subscription, 300+ models, plus the Tool Gateway:",
" web search, image generation, TTS, browser automation.",
"Sign up: https://portal.nousresearch.com/manage-subscription", None)
try:
from hermes_cli.main import _model_flow_nous
_model_flow_nous(config)
except (KeyboardInterrupt, EOFError):
_info(None, "Nous Portal setup cancelled.")
except Exception as exc:
logger.debug("_model_flow_nous error during quick setup: %s", exc)
def _on_error(exc: Exception) -> None:
print_warning(f"Nous Portal setup encountered an error: {exc}")
print_info("You can try again later with: hermes model")
# The wizard's later save_config(config) must not clobber the login/model save (#4172).
_run_nous_flow(config, context="quick setup", cancel_exc=(KeyboardInterrupt, EOFError),
cancel_lines=(None, "Nous Portal setup cancelled."), print_error=_on_error)
# The wizard's later save_config(config) must not clobber the login/model save.
_reload_config_into(config)
# Step 2: Terminal Backend; Step 3: defaults for everything else.
@@ -125,9 +114,8 @@ def _run_first_time_quick_setup(config: dict, hermes_home, is_existing: bool):
setup_gateway(config)
save_config(config)
else:
# Messaging skipped — still install/start the gateway service so cron
# jobs run and platforms come alive as soon as tokens are added later
# (e.g. via `hermes import` from another machine).
# Messaging skipped — still install/start the gateway service so cron jobs run and
# platforms come alive as soon as tokens are added later (e.g. via `hermes import`).
from hermes_cli.gateway import ensure_gateway_service
ensure_gateway_service(context="setup")
@@ -143,19 +131,14 @@ def _run_first_time_quick_setup(config: dict, hermes_home, is_existing: bool):
def _print_macos_fda_tip() -> None:
"""One-time macOS onboarding tip: a single Full Disk Access grant kills
every per-folder permission prompt, permanently (issue #52010 follow-up).
Uses the same prompt-free probe as doctor's check_macos_full_disk_access
(the TCC db dir is FDA-gated but probing it never triggers a dialog).
Silent on non-macOS and when FDA is already granted or indeterminate.
"""
"""One-time macOS tip: one Full Disk Access grant kills every per-folder prompt. Same
prompt-free probe as doctor's check_macos_full_disk_access (the FDA-gated TCC dir never
triggers a dialog); silent on non-macOS and when FDA is granted or indeterminate."""
from hermes_cli.setup import _info
if sys.platform != "darwin":
return
tcc_dir = Path.home() / "Library" / "Application Support" / "com.apple.TCC"
try:
os.listdir(tcc_dir)
os.listdir(Path.home() / "Library" / "Application Support" / "com.apple.TCC")
return # already granted — nothing to teach
except PermissionError:
pass
@@ -164,40 +147,29 @@ def _print_macos_fda_tip() -> None:
_info(None, " macOS tip: silence ALL folder permission prompts with one switch —",
" System Settings → Privacy & Security → Full Disk Access → enable",
" your terminal (and Hermes.app if you use Desktop), or run:",
" open \"x-apple.systempreferences:com.apple.preference" ".security?Privacy_AllFiles\"",
" open \"x-apple.systempreferences:com.apple.preference.security?Privacy_AllFiles\"",
" The grant is permanent — it survives every Hermes update.")
def _blank_slate_minimal_toolsets(config: dict):
"""Write the minimal toolset state for a Blank Slate install.
Only ``file``, ``terminal``, ``vision`` and ``skills`` stay on: ``read_file``
can't read images (it points at ``vision_analyze``), and the always-seeded
``hermes-agent`` skill needs ``skill_view``. Two layers enforce the selection:
1. ``platform_toolsets["cli"]`` — an explicit configurable-key list the resolver
treats as authoritative (``has_explicit_config``), so defaults aren't re-expanded.
2. ``agent.disabled_toolsets`` — a global hard-suppression list applied last in
``_get_platform_tools``, overriding even the non-configurable platform-toolset
recovery that would re-add e.g. ``kanban``. Users re-enable via ``hermes tools``
(rewrites ``platform_toolsets``) or by editing ``agent.disabled_toolsets``.
"""
"""Write the minimal toolset state for a Blank Slate install: only ``file``, ``terminal``,
``vision`` (``read_file`` can't read images) and ``skills`` (the seeded ``hermes-agent`` skill
needs ``skill_view``) stay on. Two layers enforce it: ``platform_toolsets["cli"]`` (explicit,
so defaults aren't re-expanded) and ``agent.disabled_toolsets`` (hard-suppression applied last
in ``_get_platform_tools``, overriding the recovery that would re-add e.g. ``kanban``)."""
keep = {"file", "terminal", "vision", "skills"}
config.setdefault("platform_toolsets", {})["cli"] = sorted(keep)
try:
from toolsets import TOOLSETS
from hermes_cli.tools_config import CONFIGURABLE_TOOLSETS, _get_plugin_toolset_keys
all_keys = {k for k, _, _ in CONFIGURABLE_TOOLSETS}
all_keys.update(_get_plugin_toolset_keys())
# Plain TOOLSETS entries catch recovered toolsets like ``kanban``. Skip
# "hermes-*" platform composites, "includes" groupings, and posture toolsets
# (session-level picks by agent/coding_context.py — disabling them would make
# model_tools subtract terminal/read_file from the minimal surface, #57315).
# Plain TOOLSETS entries catch recovered toolsets like ``kanban``. Skip "hermes-*" platform
# composites, "includes" groupings, and posture toolsets (session-level picks by
# agent/coding_context.py — disabling them would subtract terminal/read_file).
for k, tdef in TOOLSETS.items():
if k.startswith("hermes-"):
continue
if isinstance(tdef, dict) and (tdef.get("includes") or tdef.get("posture")):
if k.startswith("hermes-") or (isinstance(tdef, dict) and (tdef.get("includes") or tdef.get("posture"))):
continue
all_keys.add(k)
@@ -209,9 +181,7 @@ def _blank_slate_minimal_toolsets(config: dict):
def _blank_slate_minimize_config(config: dict):
"""Turn OFF the optional config features (compression, memory/profile capture,
checkpoints, smart routing, auto session reset; quiet display). All opt-in
afterwards via ``hermes setup agent`` / ``hermes config set``."""
"""Turn OFF every optional config feature; all opt back in via ``hermes setup agent``."""
config.setdefault("agent", {})["max_turns"] = 90
config.setdefault("compression", {})["enabled"] = False
mem = config.setdefault("memory", {})
@@ -223,17 +193,30 @@ def _blank_slate_minimize_config(config: dict):
config.setdefault("display", {})["tool_progress"] = "all"
def _set_bundled_skills_opt_out(opt_out: bool, log_label: str, on_success=None, on_error=None) -> None:
"""Record the bundled-skills opt-out marker and sync (essential skills are always seeded);
``on_success(sync_result)`` / ``on_error(exc)`` report the outcome."""
try:
from tools.skills_sync import set_bundled_skills_opt_out, sync_skills
set_bundled_skills_opt_out(opt_out)
result = sync_skills(quiet=True)
if on_success is not None:
on_success(result)
except Exception as exc:
logger.debug("blank-slate %s error: %s", log_label, exc)
if on_error is not None:
on_error(exc)
def _run_blank_slate_setup(config: dict, hermes_home, is_existing: bool):
"""Blank Slate setup — essentials only (provider/model, file + terminal), everything
else OFF; then either finish now or walk through opting capabilities back in.
Nothing is enabled that the user did not explicitly choose."""
"""Blank Slate setup — essentials only, everything else OFF; then finish now or walk through
opting capabilities back in. Nothing is enabled that the user did not explicitly choose."""
from hermes_cli.setup import (
_blank_slate_minimal_toolsets, _blank_slate_minimize_config, _blank_slate_walkthrough, _info,
print_header, print_info, _print_setup_summary, print_success, prompt_choice, save_config,
setup_model_provider, setup_terminal_backend,
)
print()
print_header("Blank Slate Setup")
print_header("Blank Slate Setup", gap=True)
_info("Everything starts OFF. First we force-enable only what's required",
"to run an agent, then you choose whether to stop there or walk",
"through enabling more — opting in to exactly what you want.", "",
@@ -262,33 +245,25 @@ def _run_blank_slate_setup(config: dict, hermes_home, is_existing: bool):
print_info(" Compression, memory, checkpoints, smart routing: off")
# The fork: stop here, or walk through enabling things
print()
print_header("How far do you want to go?")
print_header("How far do you want to go?", gap=True)
path = prompt_choice("Your minimal agent is ready. What next?", [
"Start with everything disabled — finish now (most minimal)",
"Walk through all configurations — opt in to tools, skills, plugins, MCP",
], 0)
if path == 0:
save_config(config)
# Blank Slate means no bundled skills; record the opt-out so future
# `hermes update` runs don't re-inject them. Essential skills (the
# `hermes-agent` operating manual) are still seeded by the sync.
try:
from tools.skills_sync import set_bundled_skills_opt_out, sync_skills
set_bundled_skills_opt_out(True)
sync_skills(quiet=True)
except Exception as exc:
logger.debug("blank-slate skill opt-out error: %s", exc)
print()
print_success("Blank Slate setup complete — minimal agent ready.")
_info("Enable anything later, on demand:", " Enable tools: hermes tools",
" Seed skills: hermes skills opt-in --sync", " Add MCP servers: hermes mcp add",
" Enable plugins: hermes plugins", " Tune agent settings: hermes setup agent", None)
_print_setup_summary(config, hermes_home)
if path != 0:
_blank_slate_walkthrough(config, hermes_home)
return
_blank_slate_walkthrough(config, hermes_home)
save_config(config)
# Blank Slate means no bundled skills; record the opt-out so future `hermes update` runs
# don't re-inject them.
_set_bundled_skills_opt_out(True, "skill opt-out")
print()
print_success("Blank Slate setup complete — minimal agent ready.")
_info("Enable anything later, on demand:", " Enable tools: hermes tools", *_BLANK_SLATE_DONE_LINES,
" Enable plugins: hermes plugins", " Tune agent settings: hermes setup agent", None)
_print_setup_summary(config, hermes_home)
def _blank_slate_walkthrough(config: dict, hermes_home):
@@ -298,35 +273,28 @@ def _blank_slate_walkthrough(config: dict, hermes_home):
save_config, setup_gateway,
)
# Bundled skills — default to NONE, offer to seed all
print()
print_header("Bundled Skills")
print_header("Bundled Skills", gap=True)
print_info("Blank Slate ships with NO bundled skills by default.")
seed_skills = prompt_yes_no(
"Seed the full bundled skill catalog? (No = start with zero skills)", default=False
seed_skills = prompt_yes_no("Seed the full bundled skill catalog? (No = start with zero skills)", default=False)
def _seeded(result) -> None:
copied = len(result.get("copied", [])) if isinstance(result, dict) else 0
print_success(f"Seeded {copied} bundled skills.")
def _opted_out(_result) -> None:
_info("No skills seeded (except the essential `hermes-agent`",
"skill). A .no-bundled-skills marker keeps future",
"`hermes update` runs from re-injecting them. Opt back in any",
"time with `hermes skills opt-in --sync`.")
# Seeding first clears any stale opt-out marker; declining sets it (essential skills still seed).
_set_bundled_skills_opt_out(
not seed_skills, "skill handling", on_success=_seeded if seed_skills else _opted_out,
on_error=lambda exc: print_warning(f"Skill setup step encountered an error: {exc}"),
)
try:
from tools.skills_sync import set_bundled_skills_opt_out, sync_skills
if seed_skills:
# Make sure no stale opt-out marker blocks the seed, then sync.
set_bundled_skills_opt_out(False)
result = sync_skills(quiet=True)
copied = len(result.get("copied", [])) if isinstance(result, dict) else 0
print_success(f"Seeded {copied} bundled skills.")
else:
set_bundled_skills_opt_out(True)
# Essential skills (`hermes-agent`) are still seeded for an opted-out profile.
sync_skills(quiet=True)
_info("No skills seeded (except the essential `hermes-agent`",
"skill). A .no-bundled-skills marker keeps future",
"`hermes update` runs from re-injecting them. Opt back in any",
"time with `hermes skills opt-in --sync`.")
except Exception as exc:
logger.debug("blank-slate skill handling error: %s", exc)
print_warning(f"Skill setup step encountered an error: {exc}")
# Walk through enabling additional tools
print()
print_header("Tools")
print_header("Tools", gap=True)
_info("Pick exactly which additional toolsets to turn on.",
"(file and terminal are already on; leave the rest off if you want", " the most minimal agent.)")
if prompt_yes_no("Open the tool selector to enable more tools?", default=False):
@@ -340,21 +308,17 @@ def _blank_slate_walkthrough(config: dict, hermes_home):
else:
print_info("Keeping the minimal toolset. Add tools later with `hermes tools`.")
# Built-in plugins (off unless chosen)
print()
print_header("Plugins")
if prompt_yes_no("Review and enable built-in plugins now?", default=False):
print_info("Manage plugins with `hermes plugins list` / `hermes plugins install`.")
else:
print_info("No plugins enabled. Add later with `hermes plugins`.")
# MCP servers (off unless chosen)
print()
print_header("MCP Servers")
if prompt_yes_no("Add an MCP server now?", default=False):
print_info("Add servers with `hermes mcp add <name> --url ... | --command ...`.")
else:
print_info("No MCP servers configured. Add later with `hermes mcp add`.")
# Built-in plugins and MCP servers (off unless chosen)
for header, question, yes_msg, no_msg in (
("Plugins", "Review and enable built-in plugins now?",
"Manage plugins with `hermes plugins list` / `hermes plugins install`.",
"No plugins enabled. Add later with `hermes plugins`."),
("MCP Servers", "Add an MCP server now?",
"Add servers with `hermes mcp add <name> --url ... | --command ...`.",
"No MCP servers configured. Add later with `hermes mcp add`."),
):
print_header(header, gap=True)
print_info(yes_msg if prompt_yes_no(question, default=False) else no_msg)
# Optional messaging gateway
print()
@@ -365,8 +329,8 @@ def _blank_slate_walkthrough(config: dict, hermes_home):
print()
print_success("Blank Slate setup complete — minimal agent ready.")
_info(" Enable more tools: hermes tools", " Seed skills: hermes skills opt-in --sync",
" Add MCP servers: hermes mcp add", " Tune agent settings: hermes setup agent", None)
_info(" Enable more tools: hermes tools", *_BLANK_SLATE_DONE_LINES,
" Tune agent settings: hermes setup agent", None)
_print_setup_summary(config, hermes_home)
@@ -375,11 +339,10 @@ def _run_quick_setup(config: dict, hermes_home):
"""Quick setup — only configure items that are missing."""
from hermes_cli.setup import (
color, Colors, _info, print_header, print_info, _print_setup_summary, print_success,
_prompt_api_key, prompt_checklist, save_config,
_prompt_and_save_env_var, _prompt_api_key, prompt_checklist, save_config,
)
from hermes_cli.config import (get_missing_env_vars, get_missing_config_fields, check_config_version)
print()
print_header("Quick Setup — Missing Items Only")
print_header("Quick Setup — Missing Items Only", gap=True)
# Check what's missing
missing_required = [v for v in get_missing_env_vars(required_only=False) if v.get("is_required")]
@@ -407,13 +370,10 @@ def _run_quick_setup(config: dict, hermes_home):
_prompt_and_save_env_var(var, f" Saved {var['name']}", f" Skipped {var['name']}")
missing_tools = [v for v in missing_optional if v.get("category") == "tool"]
missing_messaging = [
v for v in missing_optional if v.get("category") == "messaging" and not v.get("advanced")
]
missing_messaging = [v for v in missing_optional if v.get("category") == "messaging" and not v.get("advanced")]
if missing_tools: # checklist, then the API-key screen for each pick
print()
print_header("Tool API Keys")
print_header("Tool API Keys", gap=True)
labels = []
for var in missing_tools:
tools = var.get("tools", [])
@@ -423,8 +383,7 @@ def _run_quick_setup(config: dict, hermes_home):
_prompt_api_key(missing_tools[idx])
if missing_messaging: # checklist, then prompt for each selected platform's vars
print()
print_header("Messaging Platforms")
print_header("Messaging Platforms", gap=True)
_info("Connect Hermes to messaging apps to chat from anywhere.",
"You can configure these later with 'hermes setup gateway'.")
# Group by platform in first-seen order; vars matching no platform are dropped.