Merge branch 'feat/ink-refactor' of github.com:NousResearch/hermes-agent into feat/ink-refactor

This commit is contained in:
Austin Pickett
2026-04-11 22:10:11 -04:00
152 changed files with 7146 additions and 2558 deletions

View File

@@ -250,9 +250,39 @@ PROVIDER_REGISTRY: Dict[str, ProviderConfig] = {
api_key_env_vars=("HF_TOKEN",),
base_url_env_var="HF_BASE_URL",
),
"xiaomi": ProviderConfig(
id="xiaomi",
name="Xiaomi MiMo",
auth_type="api_key",
inference_base_url="https://api.xiaomimimo.com/v1",
api_key_env_vars=("XIAOMI_API_KEY",),
base_url_env_var="XIAOMI_BASE_URL",
),
}
# =============================================================================
# Anthropic Key Helper
# =============================================================================
def get_anthropic_key() -> str:
"""Return the first usable Anthropic credential, or ``""``.
Checks both the ``.env`` file (via ``get_env_value``) and the process
environment (``os.getenv``). The fallback order mirrors the
``PROVIDER_REGISTRY["anthropic"].api_key_env_vars`` tuple:
ANTHROPIC_API_KEY -> ANTHROPIC_TOKEN -> CLAUDE_CODE_OAUTH_TOKEN
"""
from hermes_cli.config import get_env_value
for var in PROVIDER_REGISTRY["anthropic"].api_key_env_vars:
value = get_env_value(var) or os.getenv(var, "")
if value:
return value
return ""
# =============================================================================
# Kimi Code Endpoint Detection
# =============================================================================
@@ -908,6 +938,7 @@ def resolve_provider(
"opencode": "opencode-zen", "zen": "opencode-zen",
"qwen-portal": "qwen-oauth", "qwen-cli": "qwen-oauth", "qwen-oauth": "qwen-oauth",
"hf": "huggingface", "hugging-face": "huggingface", "huggingface-hub": "huggingface",
"mimo": "xiaomi", "xiaomi-mimo": "xiaomi",
"go": "opencode-go", "opencode-go-sub": "opencode-go",
"kilo": "kilocode", "kilo-code": "kilocode", "kilo-gateway": "kilocode",
# Local server aliases — route through the generic custom provider

View File

@@ -1,8 +1,9 @@
"""hermes claw — OpenClaw migration commands.
Usage:
hermes claw migrate # Interactive migration from ~/.openclaw
hermes claw migrate --dry-run # Preview what would be migrated
hermes claw migrate # Preview then migrate (always shows preview first)
hermes claw migrate --dry-run # Preview only, no changes
hermes claw migrate --yes # Skip confirmation prompt
hermes claw migrate --preset full --overwrite # Full migration, overwrite conflicts
hermes claw cleanup # Archive leftover OpenClaw directories
hermes claw cleanup --dry-run # Preview what would be archived
@@ -51,6 +52,41 @@ _OPENCLAW_SCRIPT_INSTALLED = (
# Known OpenClaw directory names (current + legacy)
_OPENCLAW_DIR_NAMES = (".openclaw", ".clawdbot", ".moldbot")
def _warn_if_gateway_running(auto_yes: bool) -> None:
"""Check if a Hermes gateway is running with connected platforms.
Migrating bot tokens while the gateway is polling will cause conflicts
(e.g. Telegram 409 "terminated by other getUpdates request"). Warn the
user and let them decide whether to continue.
"""
from gateway.status import get_running_pid, read_runtime_status
if not get_running_pid():
return
data = read_runtime_status() or {}
platforms = data.get("platforms") or {}
connected = [name for name, info in platforms.items()
if isinstance(info, dict) and info.get("state") == "connected"]
if not connected:
return
print()
print_error(
"Hermes gateway is running with active connections: "
+ ", ".join(connected)
)
print_info(
"Migrating bot tokens while the gateway is active will cause "
"conflicts (Telegram, Discord, and Slack only allow one active "
"session per token)."
)
print_info("Recommendation: stop the gateway first with 'hermes stop'.")
print()
if not auto_yes and not prompt_yes_no("Continue anyway?", default=False):
print_info("Migration cancelled. Stop the gateway and try again.")
sys.exit(0)
# State files commonly found in OpenClaw workspace directories that cause
# confusion after migration (the agent discovers them and writes to them)
_WORKSPACE_STATE_GLOBS = (
@@ -237,12 +273,12 @@ def _cmd_migrate(args):
# Show what we're doing
hermes_home = get_hermes_home()
auto_yes = getattr(args, "yes", False)
print()
print_header("Migration Settings")
print_info(f"Source: {source_dir}")
print_info(f"Target: {hermes_home}")
print_info(f"Preset: {preset}")
print_info(f"Mode: {'dry run (preview only)' if dry_run else 'execute'}")
print_info(f"Overwrite: {'yes' if overwrite else 'no (skip conflicts)'}")
print_info(f"Secrets: {'yes (allowlisted only)' if migrate_secrets else 'no'}")
if skill_conflict != "skip":
@@ -251,31 +287,85 @@ def _cmd_migrate(args):
print_info(f"Workspace: {workspace_target}")
print()
# For execute mode (non-dry-run), confirm unless --yes was passed
if not dry_run and not getattr(args, "yes", False):
if not prompt_yes_no("Proceed with migration?", default=True):
print_info("Migration cancelled.")
return
# Check if a gateway is running with connected platforms — migrating tokens
# while the gateway is active will cause conflicts (e.g. Telegram 409).
_warn_if_gateway_running(auto_yes)
# Ensure config.yaml exists before migration tries to read it
config_path = get_config_path()
if not config_path.exists():
save_config(load_config())
# Load and run the migration
# Load the migration module
try:
mod = _load_migration_module(script_path)
if mod is None:
print_error("Could not load migration script.")
return
except Exception as e:
print()
print_error(f"Could not load migration script: {e}")
logger.debug("OpenClaw migration error", exc_info=True)
return
selected = mod.resolve_selected_options(None, None, preset=preset)
ws_target = Path(workspace_target).resolve() if workspace_target else None
selected = mod.resolve_selected_options(None, None, preset=preset)
ws_target = Path(workspace_target).resolve() if workspace_target else None
# ── Phase 1: Always preview first ──────────────────────────
try:
preview = mod.Migrator(
source_root=source_dir.resolve(),
target_root=hermes_home.resolve(),
execute=False,
workspace_target=ws_target,
overwrite=overwrite,
migrate_secrets=migrate_secrets,
output_dir=None,
selected_options=selected,
preset_name=preset,
skill_conflict_mode=skill_conflict,
)
preview_report = preview.migrate()
except Exception as e:
print()
print_error(f"Migration preview failed: {e}")
logger.debug("OpenClaw migration preview error", exc_info=True)
return
preview_summary = preview_report.get("summary", {})
preview_count = preview_summary.get("migrated", 0)
if preview_count == 0:
print()
print_info("Nothing to migrate from OpenClaw.")
_print_migration_report(preview_report, dry_run=True)
return
print()
print_header(f"Migration Preview — {preview_count} item(s) would be imported")
print_info("No changes have been made yet. Review the list below:")
_print_migration_report(preview_report, dry_run=True)
# If --dry-run, stop here
if dry_run:
return
# ── Phase 2: Confirm and execute ───────────────────────────
print()
if not auto_yes:
if not sys.stdin.isatty():
print_info("Non-interactive session — preview only.")
print_info("To execute, re-run with: hermes claw migrate --yes")
return
if not prompt_yes_no("Proceed with migration?", default=True):
print_info("Migration cancelled.")
return
try:
migrator = mod.Migrator(
source_root=source_dir.resolve(),
target_root=hermes_home.resolve(),
execute=not dry_run,
execute=True,
workspace_target=ws_target,
overwrite=overwrite,
migrate_secrets=migrate_secrets,
@@ -292,11 +382,11 @@ def _cmd_migrate(args):
return
# Print results
_print_migration_report(report, dry_run)
_print_migration_report(report, dry_run=False)
# After successful non-dry-run migration, offer to archive the source directory
if not dry_run and report.get("summary", {}).get("migrated", 0) > 0:
_offer_source_archival(source_dir, getattr(args, "yes", False))
# After successful migration, offer to archive the source directory
if report.get("summary", {}).get("migrated", 0) > 0:
_offer_source_archival(source_dir, auto_yes)
def _offer_source_archival(source_dir: Path, auto_yes: bool = False):
@@ -330,6 +420,11 @@ def _offer_source_archival(source_dir: Path, auto_yes: bool = False):
print_info("You can always rename it back if needed.")
print()
if not auto_yes and not sys.stdin.isatty():
print_info("Non-interactive session — skipping archival.")
print_info("Run later with: hermes claw cleanup")
return
if auto_yes or prompt_yes_no(f"Archive {source_dir} now?", default=True):
try:
archive_path = _archive_directory(source_dir)
@@ -433,6 +528,9 @@ def _cmd_cleanup(args):
if dry_run:
archive_path = _archive_directory(source_dir, dry_run=True)
print_info(f"Would archive: {source_dir}{archive_path}")
elif not auto_yes and not sys.stdin.isatty():
print_info(f"Non-interactive session — would archive: {source_dir}")
print_info("To execute, re-run with: hermes claw cleanup --yes")
else:
if auto_yes or prompt_yes_no(f"Archive {source_dir}?", default=True):
try:

79
hermes_cli/cli_output.py Normal file
View File

@@ -0,0 +1,79 @@
"""Shared CLI output helpers for Hermes CLI modules.
Extracts the identical ``print_info/success/warning/error`` and ``prompt()``
functions previously duplicated across setup.py, tools_config.py,
mcp_config.py, and memory_setup.py.
"""
import getpass
import sys
from hermes_cli.colors import Colors, color
# ─── Print Helpers ────────────────────────────────────────────────────────────
def print_info(text: str) -> None:
"""Print a dim informational message."""
print(color(f" {text}", Colors.DIM))
def print_success(text: str) -> None:
"""Print a green success message with ✓ prefix."""
print(color(f"{text}", Colors.GREEN))
def print_warning(text: str) -> None:
"""Print a yellow warning message with ⚠ prefix."""
print(color(f"{text}", Colors.YELLOW))
def print_error(text: str) -> None:
"""Print a red error message with ✗ prefix."""
print(color(f"{text}", Colors.RED))
def print_header(text: str) -> None:
"""Print a bold yellow header."""
print(color(f"\n {text}", Colors.YELLOW))
# ─── Input Prompts ────────────────────────────────────────────────────────────
def prompt(
question: str,
default: str | None = None,
password: bool = False,
) -> str:
"""Prompt the user for input with optional default and password masking.
Replaces the four independent ``_prompt()`` / ``prompt()`` implementations
in setup.py, tools_config.py, mcp_config.py, and memory_setup.py.
Returns the user's input (stripped), or *default* if the user presses Enter.
Returns empty string on Ctrl-C or EOF.
"""
suffix = f" [{default}]" if default else ""
display = color(f" {question}{suffix}: ", Colors.YELLOW)
try:
if password:
value = getpass.getpass(display)
else:
value = input(display)
value = value.strip()
return value if value else (default or "")
except (KeyboardInterrupt, EOFError):
print()
return ""
def prompt_yes_no(question: str, default: bool = True) -> bool:
"""Prompt for a yes/no answer. Returns bool."""
hint = "Y/n" if default else "y/N"
answer = prompt(f"{question} ({hint})")
if not answer:
return default
return answer.lower().startswith("y")

View File

@@ -32,7 +32,6 @@ _ENV_VAR_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_EXTRA_ENV_KEYS = frozenset({
"OPENAI_API_KEY", "OPENAI_BASE_URL",
"ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN",
"AUXILIARY_VISION_MODEL",
"DISCORD_HOME_CHANNEL", "TELEGRAM_HOME_CHANNEL",
"SIGNAL_ACCOUNT", "SIGNAL_HTTP_URL",
"SIGNAL_ALLOWED_USERS", "SIGNAL_GROUP_ALLOWED_USERS",
@@ -868,6 +867,21 @@ OPTIONAL_ENV_VARS = {
"category": "provider",
"advanced": True,
},
"XIAOMI_API_KEY": {
"description": "Xiaomi MiMo API key for MiMo models (mimo-v2-pro, mimo-v2-omni, mimo-v2-flash)",
"prompt": "Xiaomi MiMo API Key",
"url": "https://platform.xiaomimimo.com",
"password": True,
"category": "provider",
},
"XIAOMI_BASE_URL": {
"description": "Xiaomi MiMo base URL override (default: https://api.xiaomimimo.com/v1)",
"prompt": "Xiaomi base URL (leave empty for default)",
"url": None,
"password": False,
"category": "provider",
"advanced": True,
},
# ── Tool API keys ──
"EXA_API_KEY": {
@@ -1483,7 +1497,7 @@ _KNOWN_ROOT_KEYS = {
# Valid fields inside a custom_providers list entry
_VALID_CUSTOM_PROVIDER_FIELDS = {
"name", "base_url", "api_key", "api_mode", "models",
"name", "base_url", "api_key", "api_mode", "model", "models",
"context_length", "rate_limit_delay",
}
@@ -2568,7 +2582,8 @@ def show_config():
for env_key, name in keys:
value = get_env_value(env_key)
print(f" {name:<14} {redact_key(value)}")
anthropic_value = get_env_value("ANTHROPIC_TOKEN") or get_env_value("ANTHROPIC_API_KEY")
from hermes_cli.auth import get_anthropic_key
anthropic_value = get_anthropic_key()
print(f" {'Anthropic':<14} {redact_key(anthropic_value)}")
# Model settings
@@ -2784,8 +2799,8 @@ def set_config_value(key: str, value: str):
# Write only user config back (not the full merged defaults)
ensure_hermes_home()
with open(config_path, 'w', encoding="utf-8") as f:
yaml.dump(user_config, f, default_flow_style=False, sort_keys=False)
from utils import atomic_yaml_write
atomic_yaml_write(config_path, user_config, sort_keys=False)
# Keep .env in sync for keys that terminal_tool reads directly from env vars.
# config.yaml is authoritative, but terminal_tool only reads TERMINAL_ENV etc.

View File

@@ -51,6 +51,7 @@ _PROVIDER_ENV_HINTS = (
"AI_GATEWAY_API_KEY",
"OPENCODE_ZEN_API_KEY",
"OPENCODE_GO_API_KEY",
"XIAOMI_API_KEY",
)
@@ -335,8 +336,8 @@ def run_doctor(args):
model_section[k] = raw_config.pop(k)
else:
raw_config.pop(k)
with open(config_path, "w") as f:
yaml.dump(raw_config, f, default_flow_style=False)
from utils import atomic_yaml_write
atomic_yaml_write(config_path, raw_config)
check_ok("Migrated stale root-level keys into model section")
fixed_count += 1
else:
@@ -685,7 +686,8 @@ def run_doctor(args):
else:
check_warn("OpenRouter API", "(not configured)")
anthropic_key = os.getenv("ANTHROPIC_TOKEN") or os.getenv("ANTHROPIC_API_KEY")
from hermes_cli.auth import get_anthropic_key
anthropic_key = get_anthropic_key()
if anthropic_key:
print(" Checking Anthropic API...", end="", flush=True)
try:

View File

@@ -157,30 +157,54 @@ def _request_gateway_self_restart(pid: int) -> bool:
return True
def find_gateway_pids(exclude_pids: set | None = None) -> list:
def find_gateway_pids(exclude_pids: set | None = None, all_profiles: bool = False) -> list:
"""Find PIDs of running gateway processes.
Args:
exclude_pids: PIDs to exclude from the result (e.g. service-managed
PIDs that should not be killed during a stale-process sweep).
all_profiles: When ``True``, return gateway PIDs across **all**
profiles (the pre-7923 global behaviour). ``hermes update``
needs this because a code update affects every profile.
When ``False`` (default), only PIDs belonging to the current
Hermes profile are returned.
"""
pids = []
_exclude = exclude_pids or set()
pids = [pid for pid in _get_service_pids() if pid not in _exclude]
patterns = [
"hermes_cli.main gateway",
"hermes_cli.main --profile",
"hermes_cli.main -p",
"hermes_cli/main.py gateway",
"hermes_cli/main.py --profile",
"hermes_cli/main.py -p",
"hermes gateway",
"gateway/run.py",
]
current_home = str(get_hermes_home().resolve())
current_profile_arg = _profile_arg(current_home)
current_profile_name = current_profile_arg.split()[-1] if current_profile_arg else ""
def _matches_current_profile(command: str) -> bool:
if current_profile_name:
return (
f"--profile {current_profile_name}" in command
or f"-p {current_profile_name}" in command
or f"HERMES_HOME={current_home}" in command
)
if "--profile " in command or " -p " in command:
return False
if "HERMES_HOME=" in command and f"HERMES_HOME={current_home}" not in command:
return False
return True
try:
if is_windows():
# Windows: use wmic to search command lines
result = subprocess.run(
["wmic", "process", "get", "ProcessId,CommandLine", "/FORMAT:LIST"],
capture_output=True, text=True, timeout=10
)
# Parse WMIC LIST output: blocks of "CommandLine=...\nProcessId=...\n"
current_cmd = ""
for line in result.stdout.split('\n'):
line = line.strip()
@@ -188,7 +212,7 @@ def find_gateway_pids(exclude_pids: set | None = None) -> list:
current_cmd = line[len("CommandLine="):]
elif line.startswith("ProcessId="):
pid_str = line[len("ProcessId="):]
if any(p in current_cmd for p in patterns):
if any(p in current_cmd for p in patterns) and (all_profiles or _matches_current_profile(current_cmd)):
try:
pid = int(pid_str)
if pid != os.getpid() and pid not in pids and pid not in _exclude:
@@ -198,41 +222,57 @@ def find_gateway_pids(exclude_pids: set | None = None) -> list:
current_cmd = ""
else:
result = subprocess.run(
["ps", "aux"],
["ps", "eww", "-ax", "-o", "pid=,command="],
capture_output=True,
text=True,
timeout=10,
)
for line in result.stdout.split('\n'):
# Skip grep and current process
if 'grep' in line or str(os.getpid()) in line:
stripped = line.strip()
if not stripped or 'grep' in stripped:
continue
for pattern in patterns:
if pattern in line:
parts = line.split()
if len(parts) > 1:
try:
pid = int(parts[1])
if pid not in pids and pid not in _exclude:
pids.append(pid)
except ValueError:
continue
break
except Exception:
pid = None
command = ""
parts = stripped.split(None, 1)
if len(parts) == 2:
try:
pid = int(parts[0])
command = parts[1]
except ValueError:
pid = None
if pid is None:
aux_parts = stripped.split()
if len(aux_parts) > 10 and aux_parts[1].isdigit():
pid = int(aux_parts[1])
command = " ".join(aux_parts[10:])
if pid is None:
continue
if pid == os.getpid() or pid in pids or pid in _exclude:
continue
if any(pattern in command for pattern in patterns) and (all_profiles or _matches_current_profile(command)):
pids.append(pid)
except (OSError, subprocess.TimeoutExpired):
pass
return pids
def kill_gateway_processes(force: bool = False, exclude_pids: set | None = None) -> int:
def kill_gateway_processes(force: bool = False, exclude_pids: set | None = None,
all_profiles: bool = False) -> int:
"""Kill any running gateway processes. Returns count killed.
Args:
force: Use the platform's force-kill mechanism instead of graceful terminate.
exclude_pids: PIDs to skip (e.g. service-managed PIDs that were just
restarted and should not be killed).
all_profiles: When ``True``, kill across all profiles. Passed
through to :func:`find_gateway_pids`.
"""
pids = find_gateway_pids(exclude_pids=exclude_pids)
pids = find_gateway_pids(exclude_pids=exclude_pids, all_profiles=all_profiles)
killed = 0
for pid in pids:
@@ -633,6 +673,17 @@ def print_systemd_linger_guidance() -> None:
print(" If you want the gateway user service to survive logout, run:")
print(" sudo loginctl enable-linger $USER")
def _launchd_user_home() -> Path:
"""Return the real macOS user home for launchd artifacts.
Profile-mode Hermes often sets ``HOME`` to a profile-scoped directory, but
launchd user agents still live under the actual account home.
"""
import pwd
return Path(pwd.getpwuid(os.getuid()).pw_dir)
def get_launchd_plist_path() -> Path:
"""Return the launchd plist path, scoped per profile.
@@ -641,7 +692,7 @@ def get_launchd_plist_path() -> Path:
"""
suffix = _profile_suffix()
name = f"ai.hermes.gateway-{suffix}" if suffix else "ai.hermes.gateway"
return Path.home() / "Library" / "LaunchAgents" / f"{name}.plist"
return _launchd_user_home() / "Library" / "LaunchAgents" / f"{name}.plist"
def _detect_venv_dir() -> Path | None:
"""Detect the active virtualenv directory.
@@ -839,6 +890,25 @@ def _normalize_service_definition(text: str) -> str:
return "\n".join(line.rstrip() for line in text.strip().splitlines())
def _normalize_launchd_plist_for_comparison(text: str) -> str:
"""Normalize launchd plist text for staleness checks.
The generated plist intentionally captures a broad PATH assembled from the
invoking shell so user-installed tools remain reachable under launchd.
That makes raw text comparison unstable across shells, so ignore the PATH
payload when deciding whether the installed plist is stale.
"""
import re
normalized = _normalize_service_definition(text)
return re.sub(
r'(<key>PATH</key>\s*<string>)(.*?)(</string>)',
r'\1__HERMES_PATH__\3',
normalized,
flags=re.S,
)
def systemd_unit_is_current(system: bool = False) -> bool:
unit_path = get_systemd_unit_path(system=system)
if not unit_path.exists():
@@ -1220,7 +1290,7 @@ def launchd_plist_is_current() -> bool:
installed = plist_path.read_text(encoding="utf-8")
expected = generate_launchd_plist()
return _normalize_service_definition(installed) == _normalize_service_definition(expected)
return _normalize_launchd_plist_for_comparison(installed) == _normalize_launchd_plist_for_comparison(expected)
def refresh_launchd_plist_if_needed() -> bool:
@@ -1981,6 +2051,36 @@ def _setup_whatsapp():
cmd_whatsapp(argparse.Namespace())
def _setup_email():
"""Configure Email via the standard platform setup."""
email_platform = next(p for p in _PLATFORMS if p["key"] == "email")
_setup_standard_platform(email_platform)
def _setup_sms():
"""Configure SMS (Twilio) via the standard platform setup."""
sms_platform = next(p for p in _PLATFORMS if p["key"] == "sms")
_setup_standard_platform(sms_platform)
def _setup_dingtalk():
"""Configure DingTalk via the standard platform setup."""
dingtalk_platform = next(p for p in _PLATFORMS if p["key"] == "dingtalk")
_setup_standard_platform(dingtalk_platform)
def _setup_feishu():
"""Configure Feishu / Lark via the standard platform setup."""
feishu_platform = next(p for p in _PLATFORMS if p["key"] == "feishu")
_setup_standard_platform(feishu_platform)
def _setup_wecom():
"""Configure WeCom (Enterprise WeChat) via the standard platform setup."""
wecom_platform = next(p for p in _PLATFORMS if p["key"] == "wecom")
_setup_standard_platform(wecom_platform)
def _is_service_installed() -> bool:
"""Check if the gateway is installed as a system service."""
if supports_systemd_services():
@@ -2540,7 +2640,7 @@ def gateway_command(args):
service_available = True
except subprocess.CalledProcessError:
pass
killed = kill_gateway_processes()
killed = kill_gateway_processes(all_profiles=True)
total = killed + (1 if service_available else 0)
if total:
print(f"✓ Stopped {total} gateway process(es) across all profiles")

View File

@@ -606,18 +606,58 @@ def _print_tui_exit_summary(session_id: Optional[str]) -> None:
)
def _find_bundled_tui() -> Optional[Path]:
"""Find a bundled copy of the TUI.
Does *not* read from the `npm run build` dist dir,
as this would be a footgun when developing
"""
bundled_tui_dir = os.environ.get("HERMES_TUI_DIR")
if bundled_tui_dir and (Path(bundled_tui_dir) / "dist" / "entry.js").exists():
return Path(bundled_tui_dir)
def _find_bundled_tui(tui_dir: Path) -> Optional[Path]:
"""Directory whose dist/entry.js we should run: HERMES_TUI_DIR first, else repo ui-tui."""
env = os.environ.get("HERMES_TUI_DIR")
if env:
p = Path(env)
if (p / "dist" / "entry.js").exists():
return p
if (tui_dir / "dist" / "entry.js").exists():
return tui_dir
return None
def _make_tui_argv(tui_dir: Path) -> tuple[list[str], Path]:
"""Gets argv to run tui + the working directory. Will npm install deps in dev mode."""
def _tui_build_needed(tui_dir: Path) -> bool:
entry = tui_dir / "dist" / "entry.js"
if not entry.exists():
return True
dist_m = entry.stat().st_mtime
skip = frozenset({"node_modules", "dist"})
for dirpath, dirnames, filenames in os.walk(tui_dir, topdown=True):
dirnames[:] = [d for d in dirnames if d not in skip]
for fn in filenames:
if fn.endswith((".ts", ".tsx")):
if os.path.getmtime(os.path.join(dirpath, fn)) > dist_m:
return True
for meta in ("package.json", "package-lock.json", "tsconfig.json", "tsconfig.build.json"):
mp = tui_dir / meta
if mp.exists() and mp.stat().st_mtime > dist_m:
return True
return False
def _hermes_ink_bundle_stale(tui_dir: Path) -> bool:
ink_root = tui_dir / "packages" / "hermes-ink"
bundle = ink_root / "dist" / "ink-bundle.js"
if not bundle.exists():
return True
bm = bundle.stat().st_mtime
skip = frozenset({"node_modules", "dist"})
for dirpath, dirnames, filenames in os.walk(ink_root, topdown=True):
dirnames[:] = [d for d in dirnames if d not in skip]
for fn in filenames:
if fn.endswith((".ts", ".tsx")):
if os.path.getmtime(os.path.join(dirpath, fn)) > bm:
return True
mp = ink_root / "package.json"
if mp.exists() and mp.stat().st_mtime > bm:
return True
return False
def _make_tui_argv(tui_dir: Path, tui_dev: bool) -> tuple[list[str], Path]:
"""Ink TUI: --dev → tsx src; else node dist (HERMES_TUI_DIR or ui-tui, build when stale)."""
def _node_bin(bin: str)-> str:
path = shutil.which(bin)
if not path:
@@ -625,17 +665,15 @@ def _make_tui_argv(tui_dir: Path) -> tuple[list[str], Path]:
sys.exit(1)
return path
# use prebuilt TUI if it exists
bundled = _find_bundled_tui()
if bundled:
node = _node_bin("node")
return [node, str(bundled / "dist" / "entry.js")], bundled
# pre-built dist (nix / HERMES_TUI_DIR) needs no npm at all.
if not tui_dev:
bundled = _find_bundled_tui(tui_dir)
if bundled:
node = _node_bin("node")
return [node, str(bundled / "dist" / "entry.js")], bundled
# dev mode - run via tsx
# install deps if needed
npm = _node_bin("npm")
if not (tui_dir / "node_modules").exists():
npm = _node_bin("npm")
print("Installing TUI dependencies…")
result = subprocess.run(
[npm, "install", "--silent", "--no-fund", "--no-audit", "--progress=false"],
@@ -652,23 +690,60 @@ def _make_tui_argv(tui_dir: Path) -> tuple[list[str], Path]:
print(preview)
sys.exit(1)
tsx = tui_dir / "node_modules" / ".bin" / "tsx"
if tsx.exists():
return [str(tsx), "src/entry.tsx"], tui_dir
if tui_dev:
if _hermes_ink_bundle_stale(tui_dir):
result = subprocess.run(
[npm, "run", "build", "--prefix", "packages/hermes-ink"],
cwd=str(tui_dir),
capture_output=True,
text=True,
)
if result.returncode != 0:
combined = f"{result.stdout or ''}{result.stderr or ''}".strip()
preview = "\n".join(combined.splitlines()[-30:])
print("@hermes/ink build failed.")
if preview:
print(preview)
sys.exit(1)
tsx = tui_dir / "node_modules" / ".bin" / "tsx"
if tsx.exists():
return [str(tsx), "src/entry.tsx"], tui_dir
return [npm, "start"], tui_dir
npm = _node_bin("npm")
return [npm, "start"], tui_dir
if _tui_build_needed(tui_dir):
result = subprocess.run(
[npm, "run", "build"],
cwd=str(tui_dir),
capture_output=True,
text=True,
)
if result.returncode != 0:
combined = f"{result.stdout or ''}{result.stderr or ''}".strip()
preview = "\n".join(combined.splitlines()[-30:])
print("TUI build failed.")
if preview:
print(preview)
sys.exit(1)
def _launch_tui(resume_session_id: Optional[str] = None):
root = _find_bundled_tui(tui_dir)
if not root:
print("TUI build did not produce dist/entry.js")
sys.exit(1)
node = _node_bin("node")
return [node, str(root / "dist" / "entry.js")], root
def _launch_tui(resume_session_id: Optional[str] = None, tui_dev: bool = False):
"""Replace current process with the Ink TUI."""
tui_dir = PROJECT_ROOT / "ui-tui"
env = os.environ.copy()
env["HERMES_ROOT"] = os.environ.get("HERMES_ROOT", str(PROJECT_ROOT))
env["HERMES_PYTHON_SRC_ROOT"] = os.environ.get("HERMES_PYTHON_SRC_ROOT", str(PROJECT_ROOT))
env.setdefault("HERMES_CWD", os.getcwd())
if resume_session_id:
env["HERMES_TUI_RESUME"] = resume_session_id
argv, cwd = _make_tui_argv(tui_dir)
argv, cwd = _make_tui_argv(tui_dir, tui_dev)
try:
code = subprocess.call(argv, cwd=str(cwd), env=env)
except KeyboardInterrupt:
@@ -718,9 +793,6 @@ def cmd_chat(args):
# If resolution fails, keep the original value — _init_agent will
# report "Session not found" with the original input
if use_tui:
_launch_tui(getattr(args, "resume", None))
# First-run guard: check if any provider is configured before launching
if not _has_any_provider_configured():
print()
@@ -770,6 +842,13 @@ def cmd_chat(args):
if getattr(args, "source", None):
os.environ["HERMES_SESSION_SOURCE"] = args.source
if use_tui:
_launch_tui(
getattr(args, "resume", None),
tui_dev=getattr(args, "tui_dev", False),
)
# Import and run the CLI
from cli import main as cli_main
@@ -1069,6 +1148,7 @@ def select_provider_and_model(args=None):
"kilocode": "Kilo Code",
"alibaba": "Alibaba Cloud (DashScope)",
"huggingface": "Hugging Face",
"xiaomi": "Xiaomi MiMo",
"custom": "Custom endpoint",
}
active_label = provider_labels.get(active, active) if active else "none"
@@ -1101,6 +1181,7 @@ def select_provider_and_model(args=None):
("opencode-go", "OpenCode Go (open models, $10/month subscription)"),
("ai-gateway", "AI Gateway (Vercel — 200+ models, pay-per-use)"),
("alibaba", "Alibaba Cloud / DashScope Coding (Qwen + multi-provider)"),
("xiaomi", "Xiaomi MiMo (MiMo-V2 models — pro, omni, flash)"),
]
def _named_custom_provider_map(cfg) -> dict[str, dict[str, str]]:
@@ -1212,7 +1293,7 @@ def select_provider_and_model(args=None):
_model_flow_anthropic(config, current_model)
elif selected_provider == "kimi-coding":
_model_flow_kimi(config, current_model)
elif selected_provider in ("gemini", "zai", "minimax", "minimax-cn", "kilocode", "opencode-zen", "opencode-go", "ai-gateway", "alibaba", "huggingface"):
elif selected_provider in ("gemini", "zai", "minimax", "minimax-cn", "kilocode", "opencode-zen", "opencode-go", "ai-gateway", "alibaba", "huggingface", "xiaomi"):
_model_flow_api_key_provider(config, selected_provider, current_model)
# ── Post-switch cleanup: clear stale OPENAI_BASE_URL ──────────────
@@ -2682,13 +2763,8 @@ def _model_flow_anthropic(config, current_model=""):
from hermes_cli.models import _PROVIDER_MODELS
# Check ALL credential sources
existing_key = (
get_env_value("ANTHROPIC_TOKEN")
or os.getenv("ANTHROPIC_TOKEN", "")
or get_env_value("ANTHROPIC_API_KEY")
or os.getenv("ANTHROPIC_API_KEY", "")
or os.getenv("CLAUDE_CODE_OAUTH_TOKEN", "")
)
from hermes_cli.auth import get_anthropic_key
existing_key = get_anthropic_key()
cc_available = False
try:
from agent.anthropic_adapter import read_claude_code_credentials, is_claude_code_token_valid
@@ -3062,6 +3138,8 @@ def _update_via_zip(args):
)
_install_python_dependencies_with_optional_fallback(pip_cmd)
_update_node_dependencies()
# Sync skills
try:
from tools.skills_sync import sync_skills
@@ -3581,9 +3659,42 @@ def _install_python_dependencies_with_optional_fallback(
print(f" ⚠ Skipped optional extras that still failed: {', '.join(failed_extras)}")
def _update_node_dependencies() -> None:
npm = shutil.which("npm")
if not npm:
return
paths = (
("repo root", PROJECT_ROOT),
("ui-tui", PROJECT_ROOT / "ui-tui"),
)
if not any((path / "package.json").exists() for _, path in paths):
return
print("→ Updating Node.js dependencies...")
for label, path in paths:
if not (path / "package.json").exists():
continue
result = subprocess.run(
[npm, "install", "--silent", "--no-fund", "--no-audit", "--progress=false"],
cwd=path,
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
print(f"{label}")
continue
print(f" ⚠ npm install failed in {label}")
stderr = (result.stderr or "").strip()
if stderr:
print(f" {stderr.splitlines()[-1]}")
def cmd_update(args):
"""Update Hermes Agent to the latest version."""
import shutil
from hermes_cli.config import is_managed, managed_error
if is_managed():
@@ -3802,13 +3913,8 @@ def cmd_update(args):
)
_install_python_dependencies_with_optional_fallback(pip_cmd)
# Check for Node.js deps
if (PROJECT_ROOT / "package.json").exists():
import shutil
if shutil.which("npm"):
print("→ Updating Node.js dependencies...")
subprocess.run(["npm", "install", "--silent"], cwd=PROJECT_ROOT, check=False)
_update_node_dependencies()
print()
print("✓ Code updated!")
@@ -4014,7 +4120,7 @@ def cmd_update(args):
# Exclude PIDs that belong to just-restarted services so we don't
# immediately kill the process that systemd/launchd just spawned.
service_pids = _get_service_pids()
manual_pids = find_gateway_pids(exclude_pids=service_pids)
manual_pids = find_gateway_pids(exclude_pids=service_pids, all_profiles=True)
for pid in manual_pids:
try:
os.kill(pid, _signal.SIGTERM)
@@ -4463,7 +4569,14 @@ For more help on a command:
default=False,
help="Launch the Ink-based terminal UI instead of the classic REPL"
)
parser.add_argument(
"--dev",
dest="tui_dev",
action="store_true",
default=False,
help="With --tui: run TypeScript sources via tsx (skip dist build)",
)
subparsers = parser.add_subparsers(dest="command", help="Command to run")
# =========================================================================
@@ -4498,7 +4611,7 @@ For more help on a command:
)
chat_parser.add_argument(
"--provider",
choices=["auto", "openrouter", "nous", "openai-codex", "copilot-acp", "copilot", "anthropic", "gemini", "huggingface", "zai", "kimi-coding", "minimax", "minimax-cn", "kilocode"],
choices=["auto", "openrouter", "nous", "openai-codex", "copilot-acp", "copilot", "anthropic", "gemini", "huggingface", "zai", "kimi-coding", "minimax", "minimax-cn", "kilocode", "xiaomi"],
default=None,
help="Inference provider (default: auto)"
)
@@ -4569,6 +4682,13 @@ For more help on a command:
default=False,
help="Launch the Ink-based terminal UI instead of the classic REPL"
)
chat_parser.add_argument(
"--dev",
dest="tui_dev",
action="store_true",
default=False,
help="With --tui: run TypeScript sources via tsx (skip dist build)",
)
chat_parser.set_defaults(func=cmd_chat)
# =========================================================================
@@ -5558,7 +5678,8 @@ For more help on a command:
claw_migrate = claw_subparsers.add_parser(
"migrate",
help="Migrate from OpenClaw to Hermes",
description="Import settings, memories, skills, and API keys from an OpenClaw installation"
description="Import settings, memories, skills, and API keys from an OpenClaw installation. "
"Always shows a preview before making changes."
)
claw_migrate.add_argument(
"--source",
@@ -5567,7 +5688,7 @@ For more help on a command:
claw_migrate.add_argument(
"--dry-run",
action="store_true",
help="Preview what would be migrated without making changes"
help="Preview only — stop after showing what would be migrated"
)
claw_migrate.add_argument(
"--preset",

View File

@@ -57,19 +57,8 @@ def _confirm(question: str, default: bool = True) -> bool:
def _prompt(question: str, *, password: bool = False, default: str = "") -> str:
display = f" {question}"
if default:
display += f" [{default}]"
display += ": "
try:
if password:
value = getpass.getpass(color(display, Colors.YELLOW))
else:
value = input(color(display, Colors.YELLOW))
return value.strip() or default
except (KeyboardInterrupt, EOFError):
print()
return default
from hermes_cli.cli_output import prompt as _shared_prompt
return _shared_prompt(question, default=default, password=password)
# ─── Config Helpers ───────────────────────────────────────────────────────────

View File

@@ -25,85 +25,13 @@ def _curses_select(title: str, items: list[tuple[str, str]], default: int = 0) -
items: list of (label, description) tuples.
Returns selected index, or default on escape/quit.
"""
try:
import curses
result = [default]
def _menu(stdscr):
curses.curs_set(0)
if curses.has_colors():
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_GREEN, -1)
curses.init_pair(2, curses.COLOR_YELLOW, -1)
curses.init_pair(3, curses.COLOR_CYAN, -1)
cursor = default
while True:
stdscr.clear()
max_y, max_x = stdscr.getmaxyx()
# Title
try:
stdscr.addnstr(0, 0, title, max_x - 1,
curses.A_BOLD | (curses.color_pair(2) if curses.has_colors() else 0))
stdscr.addnstr(1, 0, " ↑↓ navigate ⏎ select q quit", max_x - 1,
curses.color_pair(3) if curses.has_colors() else curses.A_DIM)
except curses.error:
pass
for i, (label, desc) in enumerate(items):
y = i + 3
if y >= max_y - 1:
break
arrow = "" if i == cursor else " "
line = f" {arrow} {label}"
if desc:
line += f" {desc}"
attr = curses.A_NORMAL
if i == cursor:
attr = curses.A_BOLD
if curses.has_colors():
attr |= curses.color_pair(1)
try:
stdscr.addnstr(y, 0, line[:max_x - 1], max_x - 1, attr)
except curses.error:
pass
stdscr.refresh()
key = stdscr.getch()
if key in (curses.KEY_UP, ord('k')):
cursor = (cursor - 1) % len(items)
elif key in (curses.KEY_DOWN, ord('j')):
cursor = (cursor + 1) % len(items)
elif key in (curses.KEY_ENTER, 10, 13):
result[0] = cursor
return
elif key in (27, ord('q')):
return
curses.wrapper(_menu)
return result[0]
except Exception:
# Fallback: numbered input
print(f"\n {title}\n")
for i, (label, desc) in enumerate(items):
marker = "" if i == default else " "
d = f" {desc}" if desc else ""
print(f" {marker} {i + 1}. {label}{d}")
while True:
try:
val = input(f"\n Select [1-{len(items)}] ({default + 1}): ")
if not val:
return default
idx = int(val) - 1
if 0 <= idx < len(items):
return idx
except (ValueError, EOFError):
return default
from hermes_cli.curses_ui import curses_radiolist
# Format (label, desc) tuples into display strings
display_items = [
f"{label} {desc}" if desc else label
for label, desc in items
]
return curses_radiolist(title, display_items, selected=default, cancel_returns=default)
def _prompt(label: str, default: str | None = None, secret: bool = False) -> str:

View File

@@ -92,6 +92,7 @@ _MATCHING_PREFIX_STRIP_PROVIDERS: frozenset[str] = frozenset({
"minimax-cn",
"alibaba",
"qwen-oauth",
"xiaomi",
"custom",
})

View File

@@ -188,6 +188,11 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"deepseek-chat",
"deepseek-reasoner",
],
"xiaomi": [
"mimo-v2-pro",
"mimo-v2-omni",
"mimo-v2-flash",
],
"opencode-zen": [
"gpt-5.4-pro",
"gpt-5.4",
@@ -493,6 +498,7 @@ _PROVIDER_LABELS = {
"alibaba": "Alibaba Cloud (DashScope)",
"qwen-oauth": "Qwen OAuth (Portal)",
"huggingface": "Hugging Face",
"xiaomi": "Xiaomi MiMo",
"custom": "Custom endpoint",
}
@@ -535,6 +541,8 @@ _PROVIDER_ALIASES = {
"hf": "huggingface",
"hugging-face": "huggingface",
"huggingface-hub": "huggingface",
"mimo": "xiaomi",
"xiaomi-mimo": "xiaomi",
}
@@ -819,7 +827,7 @@ def list_available_providers() -> list[dict[str, str]]:
"openrouter", "nous", "openai-codex", "copilot", "copilot-acp",
"gemini", "huggingface",
"zai", "kimi-coding", "minimax", "minimax-cn", "kilocode", "anthropic", "alibaba",
"qwen-oauth",
"qwen-oauth", "xiaomi",
"opencode-zen", "opencode-go",
"ai-gateway", "deepseek", "custom",
]

45
hermes_cli/platforms.py Normal file
View File

@@ -0,0 +1,45 @@
"""
Shared platform registry for Hermes Agent.
Single source of truth for platform metadata consumed by both
skills_config (label display) and tools_config (default toolset
resolution). Import ``PLATFORMS`` from here instead of maintaining
duplicate dicts in each module.
"""
from collections import OrderedDict
from typing import NamedTuple
class PlatformInfo(NamedTuple):
"""Metadata for a single platform entry."""
label: str
default_toolset: str
# Ordered so that TUI menus are deterministic.
PLATFORMS: OrderedDict[str, PlatformInfo] = OrderedDict([
("cli", PlatformInfo(label="🖥️ CLI", default_toolset="hermes-cli")),
("telegram", PlatformInfo(label="📱 Telegram", default_toolset="hermes-telegram")),
("discord", PlatformInfo(label="💬 Discord", default_toolset="hermes-discord")),
("slack", PlatformInfo(label="💼 Slack", default_toolset="hermes-slack")),
("whatsapp", PlatformInfo(label="📱 WhatsApp", default_toolset="hermes-whatsapp")),
("signal", PlatformInfo(label="📡 Signal", default_toolset="hermes-signal")),
("bluebubbles", PlatformInfo(label="💙 BlueBubbles", default_toolset="hermes-bluebubbles")),
("email", PlatformInfo(label="📧 Email", default_toolset="hermes-email")),
("homeassistant", PlatformInfo(label="🏠 Home Assistant", default_toolset="hermes-homeassistant")),
("mattermost", PlatformInfo(label="💬 Mattermost", default_toolset="hermes-mattermost")),
("matrix", PlatformInfo(label="💬 Matrix", default_toolset="hermes-matrix")),
("dingtalk", PlatformInfo(label="💬 DingTalk", default_toolset="hermes-dingtalk")),
("feishu", PlatformInfo(label="🪽 Feishu", default_toolset="hermes-feishu")),
("wecom", PlatformInfo(label="💬 WeCom", default_toolset="hermes-wecom")),
("weixin", PlatformInfo(label="💬 Weixin", default_toolset="hermes-weixin")),
("webhook", PlatformInfo(label="🔗 Webhook", default_toolset="hermes-webhook")),
("api_server", PlatformInfo(label="🌐 API Server", default_toolset="hermes-api-server")),
])
def platform_label(key: str, default: str = "") -> str:
"""Return the display label for a platform key, or *default*."""
info = PLATFORMS.get(key)
return info.label if info is not None else default

View File

@@ -132,6 +132,10 @@ HERMES_OVERLAYS: Dict[str, HermesOverlay] = {
base_url_override="https://api.x.ai/v1",
base_url_env_var="XAI_BASE_URL",
),
"xiaomi": HermesOverlay(
transport="openai_chat",
base_url_env_var="XIAOMI_BASE_URL",
),
}
@@ -222,6 +226,10 @@ ALIASES: Dict[str, str] = {
"hugging-face": "huggingface",
"huggingface-hub": "huggingface",
# xiaomi
"mimo": "xiaomi",
"xiaomi-mimo": "xiaomi",
# Local server aliases → virtual "local" concept (resolved via user config)
"lmstudio": "lmstudio",
"lm-studio": "lmstudio",
@@ -242,6 +250,7 @@ _LABEL_OVERRIDES: Dict[str, str] = {
"nous": "Nous Portal",
"openai-codex": "OpenAI Codex",
"copilot-acp": "GitHub Copilot ACP",
"xiaomi": "Xiaomi MiMo",
"local": "Local endpoint",
}

View File

@@ -304,6 +304,9 @@ def _get_named_custom_provider(requested_provider: str) -> Optional[Dict[str, An
api_mode = _parse_api_mode(entry.get("api_mode"))
if api_mode:
result["api_mode"] = api_mode
model_name = str(entry.get("model", "") or "").strip()
if model_name:
result["model"] = model_name
return result
return None
@@ -329,6 +332,11 @@ def _resolve_named_custom_runtime(
# Check if a credential pool exists for this custom endpoint
pool_result = _try_resolve_from_custom_pool(base_url, "custom", custom_provider.get("api_mode"))
if pool_result:
# Propagate the model name even when using pooled credentials —
# the pool doesn't know about the custom_providers model field.
model_name = custom_provider.get("model")
if model_name:
pool_result["model"] = model_name
return pool_result
api_key_candidates = [
@@ -339,7 +347,7 @@ def _resolve_named_custom_runtime(
]
api_key = next((candidate for candidate in api_key_candidates if has_usable_secret(candidate)), "")
return {
result = {
"provider": "custom",
"api_mode": custom_provider.get("api_mode")
or _detect_api_mode_for_url(base_url)
@@ -348,6 +356,11 @@ def _resolve_named_custom_runtime(
"api_key": api_key or "no-key-required",
"source": f"custom_provider:{custom_provider.get('name', requested_provider)}",
}
# Propagate the model name so callers can override self.model when the
# provider name differs from the actual model string the API expects.
if custom_provider.get("model"):
result["model"] = custom_provider["model"]
return result
def _resolve_openrouter_runtime(

View File

@@ -197,24 +197,12 @@ def print_header(title: str):
print(color(f"{title}", Colors.CYAN, Colors.BOLD))
def print_info(text: str):
"""Print info text."""
print(color(f" {text}", Colors.DIM))
def print_success(text: str):
"""Print success message."""
print(color(f"{text}", Colors.GREEN))
def print_warning(text: str):
"""Print warning message."""
print(color(f"{text}", Colors.YELLOW))
def print_error(text: str):
"""Print error message."""
print(color(f"{text}", Colors.RED))
from hermes_cli.cli_output import ( # noqa: E402
print_error,
print_info,
print_success,
print_warning,
)
def is_interactive_stdin() -> bool:
@@ -269,80 +257,9 @@ def prompt(question: str, default: str = None, password: bool = False) -> str:
def _curses_prompt_choice(question: str, choices: list, default: int = 0) -> int:
"""Single-select menu using curses to avoid simple_term_menu rendering bugs."""
try:
import curses
result_holder = [default]
def _curses_menu(stdscr):
curses.curs_set(0)
if curses.has_colors():
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_GREEN, -1)
curses.init_pair(2, curses.COLOR_YELLOW, -1)
cursor = default
scroll_offset = 0
while True:
stdscr.clear()
max_y, max_x = stdscr.getmaxyx()
# Rows available for list items: rows 2..(max_y-2) inclusive.
visible = max(1, max_y - 3)
# Scroll the viewport so the cursor is always visible.
if cursor < scroll_offset:
scroll_offset = cursor
elif cursor >= scroll_offset + visible:
scroll_offset = cursor - visible + 1
scroll_offset = max(0, min(scroll_offset, max(0, len(choices) - visible)))
try:
stdscr.addnstr(
0,
0,
question,
max_x - 1,
curses.A_BOLD | (curses.color_pair(2) if curses.has_colors() else 0),
)
except curses.error:
pass
for row, i in enumerate(range(scroll_offset, min(scroll_offset + visible, len(choices)))):
y = row + 2
if y >= max_y - 1:
break
arrow = "" if i == cursor else " "
line = f" {arrow} {choices[i]}"
attr = curses.A_NORMAL
if i == cursor:
attr = curses.A_BOLD
if curses.has_colors():
attr |= curses.color_pair(1)
try:
stdscr.addnstr(y, 0, line, max_x - 1, attr)
except curses.error:
pass
stdscr.refresh()
key = stdscr.getch()
if key in (curses.KEY_UP, ord("k")):
cursor = (cursor - 1) % len(choices)
elif key in (curses.KEY_DOWN, ord("j")):
cursor = (cursor + 1) % len(choices)
elif key in (curses.KEY_ENTER, 10, 13):
result_holder[0] = cursor
return
elif key in (27, ord("q")):
return
curses.wrapper(_curses_menu)
from hermes_cli.curses_ui import flush_stdin
flush_stdin()
return result_holder[0]
except Exception:
return -1
"""Single-select menu using curses. Delegates to curses_radiolist."""
from hermes_cli.curses_ui import curses_radiolist
return curses_radiolist(question, choices, selected=default, cancel_returns=-1)
@@ -2052,6 +1969,42 @@ def _setup_weixin():
_gateway_setup_weixin()
def _setup_signal():
"""Configure Signal via gateway setup."""
from hermes_cli.gateway import _setup_signal as _gateway_setup_signal
_gateway_setup_signal()
def _setup_email():
"""Configure Email via gateway setup."""
from hermes_cli.gateway import _setup_email as _gateway_setup_email
_gateway_setup_email()
def _setup_sms():
"""Configure SMS (Twilio) via gateway setup."""
from hermes_cli.gateway import _setup_sms as _gateway_setup_sms
_gateway_setup_sms()
def _setup_dingtalk():
"""Configure DingTalk via gateway setup."""
from hermes_cli.gateway import _setup_dingtalk as _gateway_setup_dingtalk
_gateway_setup_dingtalk()
def _setup_feishu():
"""Configure Feishu / Lark via gateway setup."""
from hermes_cli.gateway import _setup_feishu as _gateway_setup_feishu
_gateway_setup_feishu()
def _setup_wecom():
"""Configure WeCom (Enterprise WeChat) via gateway setup."""
from hermes_cli.gateway import _setup_wecom as _gateway_setup_wecom
_gateway_setup_wecom()
def _setup_bluebubbles():
"""Configure BlueBubbles iMessage gateway."""
print_header("BlueBubbles (iMessage)")
@@ -2168,9 +2121,15 @@ _GATEWAY_PLATFORMS = [
("Telegram", "TELEGRAM_BOT_TOKEN", _setup_telegram),
("Discord", "DISCORD_BOT_TOKEN", _setup_discord),
("Slack", "SLACK_BOT_TOKEN", _setup_slack),
("Signal", "SIGNAL_HTTP_URL", _setup_signal),
("Email", "EMAIL_ADDRESS", _setup_email),
("SMS (Twilio)", "TWILIO_ACCOUNT_SID", _setup_sms),
("Matrix", "MATRIX_ACCESS_TOKEN", _setup_matrix),
("Mattermost", "MATTERMOST_TOKEN", _setup_mattermost),
("WhatsApp", "WHATSAPP_ENABLED", _setup_whatsapp),
("DingTalk", "DINGTALK_CLIENT_ID", _setup_dingtalk),
("Feishu / Lark", "FEISHU_APP_ID", _setup_feishu),
("WeCom (Enterprise WeChat)", "WECOM_BOT_ID", _setup_wecom),
("Weixin (WeChat)", "WEIXIN_ACCOUNT_ID", _setup_weixin),
("BlueBubbles (iMessage)", "BLUEBUBBLES_SERVER_URL", _setup_bluebubbles),
("Webhooks (GitHub, GitLab, etc.)", "WEBHOOK_ENABLED", _setup_webhooks),
@@ -2212,10 +2171,17 @@ def setup_gateway(config: dict):
get_env_value("TELEGRAM_BOT_TOKEN")
or get_env_value("DISCORD_BOT_TOKEN")
or get_env_value("SLACK_BOT_TOKEN")
or get_env_value("SIGNAL_HTTP_URL")
or get_env_value("EMAIL_ADDRESS")
or get_env_value("TWILIO_ACCOUNT_SID")
or get_env_value("MATTERMOST_TOKEN")
or get_env_value("MATRIX_ACCESS_TOKEN")
or get_env_value("MATRIX_PASSWORD")
or get_env_value("WHATSAPP_ENABLED")
or get_env_value("DINGTALK_CLIENT_ID")
or get_env_value("FEISHU_APP_ID")
or get_env_value("WECOM_BOT_ID")
or get_env_value("WEIXIN_ACCOUNT_ID")
or get_env_value("BLUEBUBBLES_SERVER_URL")
or get_env_value("WEBHOOK_ENABLED")
)
@@ -2404,12 +2370,30 @@ def _get_section_config_summary(config: dict, section_key: str) -> Optional[str]
platforms.append("Discord")
if get_env_value("SLACK_BOT_TOKEN"):
platforms.append("Slack")
if get_env_value("WHATSAPP_PHONE_NUMBER_ID"):
platforms.append("WhatsApp")
if get_env_value("SIGNAL_ACCOUNT"):
platforms.append("Signal")
if get_env_value("EMAIL_ADDRESS"):
platforms.append("Email")
if get_env_value("TWILIO_ACCOUNT_SID"):
platforms.append("SMS")
if get_env_value("MATRIX_ACCESS_TOKEN") or get_env_value("MATRIX_PASSWORD"):
platforms.append("Matrix")
if get_env_value("MATTERMOST_TOKEN"):
platforms.append("Mattermost")
if get_env_value("WHATSAPP_PHONE_NUMBER_ID"):
platforms.append("WhatsApp")
if get_env_value("DINGTALK_CLIENT_ID"):
platforms.append("DingTalk")
if get_env_value("FEISHU_APP_ID"):
platforms.append("Feishu")
if get_env_value("WECOM_BOT_ID"):
platforms.append("WeCom")
if get_env_value("WEIXIN_ACCOUNT_ID"):
platforms.append("Weixin")
if get_env_value("BLUEBUBBLES_SERVER_URL"):
platforms.append("BlueBubbles")
if get_env_value("WEBHOOK_ENABLED"):
platforms.append("Webhooks")
if platforms:
return ", ".join(platforms)
return None # No platforms configured — section must run

View File

@@ -15,25 +15,12 @@ from typing import List, Optional, Set
from hermes_cli.config import load_config, save_config
from hermes_cli.colors import Colors, color
from hermes_cli.platforms import PLATFORMS as _PLATFORMS, platform_label
PLATFORMS = {
"cli": "🖥️ CLI",
"telegram": "📱 Telegram",
"discord": "💬 Discord",
"slack": "💼 Slack",
"whatsapp": "📱 WhatsApp",
"signal": "📡 Signal",
"bluebubbles": "💬 BlueBubbles",
"email": "📧 Email",
"homeassistant": "🏠 Home Assistant",
"mattermost": "💬 Mattermost",
"matrix": "💬 Matrix",
"dingtalk": "💬 DingTalk",
"feishu": "🪽 Feishu",
"wecom": "💬 WeCom",
"weixin": "💬 Weixin",
"webhook": "🔗 Webhook",
}
# Backward-compatible view: {key: label_string} so existing code that
# iterates ``PLATFORMS.items()`` or calls ``PLATFORMS.get(key)`` keeps
# working without changes to every call site.
PLATFORMS = {k: info.label for k, info in _PLATFORMS.items() if k != "api_server"}
# ─── Config Helpers ───────────────────────────────────────────────────────────

View File

@@ -141,11 +141,8 @@ def show_status(args):
display = redact_key(value) if not show_all else value
print(f" {name:<12} {check_mark(has_key)} {display}")
anthropic_value = (
get_env_value("ANTHROPIC_TOKEN")
or get_env_value("ANTHROPIC_API_KEY")
or ""
)
from hermes_cli.auth import get_anthropic_key
anthropic_value = get_anthropic_key()
anthropic_display = redact_key(anthropic_value) if not show_all else anthropic_value
print(f" {'Anthropic':<12} {check_mark(bool(anthropic_value))} {anthropic_display}")

View File

@@ -33,33 +33,13 @@ PROJECT_ROOT = Path(__file__).parent.parent.resolve()
# ─── UI Helpers (shared with setup.py) ────────────────────────────────────────
def _print_info(text: str):
print(color(f" {text}", Colors.DIM))
def _print_success(text: str):
print(color(f"{text}", Colors.GREEN))
def _print_warning(text: str):
print(color(f"{text}", Colors.YELLOW))
def _print_error(text: str):
print(color(f"{text}", Colors.RED))
def _prompt(question: str, default: str = None, password: bool = False) -> str:
if default:
display = f"{question} [{default}]: "
else:
display = f"{question}: "
try:
if password:
import getpass
value = getpass.getpass(color(display, Colors.YELLOW))
else:
value = input(color(display, Colors.YELLOW))
return value.strip() or default or ""
except (KeyboardInterrupt, EOFError):
print()
return default or ""
from hermes_cli.cli_output import ( # noqa: E402 — late import block
print_error as _print_error,
print_info as _print_info,
print_success as _print_success,
print_warning as _print_warning,
prompt as _prompt,
)
# ─── Toolset Registry ─────────────────────────────────────────────────────────
@@ -118,25 +98,14 @@ def _get_plugin_toolset_keys() -> set:
except Exception:
return set()
# Platform display config
# Platform display config — derived from the canonical registry so every
# module shares the same data. Kept as dict-of-dicts for backward
# compatibility with existing ``PLATFORMS[key]["label"]`` access patterns.
from hermes_cli.platforms import PLATFORMS as _PLATFORMS_REGISTRY
PLATFORMS = {
"cli": {"label": "🖥️ CLI", "default_toolset": "hermes-cli"},
"telegram": {"label": "📱 Telegram", "default_toolset": "hermes-telegram"},
"discord": {"label": "💬 Discord", "default_toolset": "hermes-discord"},
"slack": {"label": "💼 Slack", "default_toolset": "hermes-slack"},
"whatsapp": {"label": "📱 WhatsApp", "default_toolset": "hermes-whatsapp"},
"signal": {"label": "📡 Signal", "default_toolset": "hermes-signal"},
"bluebubbles": {"label": "💙 BlueBubbles", "default_toolset": "hermes-bluebubbles"},
"homeassistant": {"label": "🏠 Home Assistant", "default_toolset": "hermes-homeassistant"},
"email": {"label": "📧 Email", "default_toolset": "hermes-email"},
"matrix": {"label": "💬 Matrix", "default_toolset": "hermes-matrix"},
"dingtalk": {"label": "💬 DingTalk", "default_toolset": "hermes-dingtalk"},
"feishu": {"label": "🪽 Feishu", "default_toolset": "hermes-feishu"},
"wecom": {"label": "💬 WeCom", "default_toolset": "hermes-wecom"},
"weixin": {"label": "💬 Weixin", "default_toolset": "hermes-weixin"},
"api_server": {"label": "🌐 API Server", "default_toolset": "hermes-api-server"},
"mattermost": {"label": "💬 Mattermost", "default_toolset": "hermes-mattermost"},
"webhook": {"label": "🔗 Webhook", "default_toolset": "hermes-webhook"},
k: {"label": info.label, "default_toolset": info.default_toolset}
for k, info in _PLATFORMS_REGISTRY.items()
}
@@ -677,86 +646,9 @@ def _toolset_has_keys(ts_key: str, config: dict = None) -> bool:
# ─── Menu Helpers ─────────────────────────────────────────────────────────────
def _prompt_choice(question: str, choices: list, default: int = 0) -> int:
"""Single-select menu (arrow keys). Uses curses to avoid simple_term_menu
rendering bugs in tmux, iTerm, and other non-standard terminals."""
# Curses-based single-select — works in tmux, iTerm, and standard terminals
try:
import curses
result_holder = [default]
def _curses_menu(stdscr):
curses.curs_set(0)
if curses.has_colors():
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_GREEN, -1)
curses.init_pair(2, curses.COLOR_YELLOW, -1)
cursor = default
while True:
stdscr.clear()
max_y, max_x = stdscr.getmaxyx()
try:
stdscr.addnstr(0, 0, question, max_x - 1,
curses.A_BOLD | (curses.color_pair(2) if curses.has_colors() else 0))
except curses.error:
pass
for i, c in enumerate(choices):
y = i + 2
if y >= max_y - 1:
break
arrow = "" if i == cursor else " "
line = f" {arrow} {c}"
attr = curses.A_NORMAL
if i == cursor:
attr = curses.A_BOLD
if curses.has_colors():
attr |= curses.color_pair(1)
try:
stdscr.addnstr(y, 0, line, max_x - 1, attr)
except curses.error:
pass
stdscr.refresh()
key = stdscr.getch()
if key in (curses.KEY_UP, ord('k')):
cursor = (cursor - 1) % len(choices)
elif key in (curses.KEY_DOWN, ord('j')):
cursor = (cursor + 1) % len(choices)
elif key in (curses.KEY_ENTER, 10, 13):
result_holder[0] = cursor
return
elif key in (27, ord('q')):
return
curses.wrapper(_curses_menu)
from hermes_cli.curses_ui import flush_stdin
flush_stdin()
return result_holder[0]
except Exception:
pass
# Fallback: numbered input (Windows without curses, etc.)
print(color(question, Colors.YELLOW))
for i, c in enumerate(choices):
marker = "" if i == default else ""
style = Colors.GREEN if i == default else ""
print(color(f" {marker} {i+1}. {c}", style) if style else f" {marker} {i+1}. {c}")
while True:
try:
val = input(color(f" Select [1-{len(choices)}] ({default + 1}): ", Colors.DIM))
if not val:
return default
idx = int(val) - 1
if 0 <= idx < len(choices):
return idx
except (ValueError, KeyboardInterrupt, EOFError):
print()
return default
"""Single-select menu (arrow keys). Delegates to curses_radiolist."""
from hermes_cli.curses_ui import curses_radiolist
return curses_radiolist(question, choices, selected=default, cancel_returns=default)
# ─── Token Estimation ────────────────────────────────────────────────────────