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

refactor(hermes_cli): tools_config — _select_into() unifies 6 provider-selection writes, gen-model picker partials, table-driven menu loop, _configure_platforms() merges one/all-platform flows (fuzz-verified byte-identical)

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:11:10 -07:00
parent ca46f2826e
commit ec4583f970
2 changed files with 97 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
hermes_cli/tools_config.py
View File

@@ -443,13 +443,6 @@ def enabled_mcp_server_names(config: dict) -> Set[str]:
return names
def _exempt_explicit_platform_native(default_off: Set[str], platform: str, *, explicitly_configured: bool) -> None:
"""Let platform-native default-off toolsets (``discord`` on discord) through on explicit config (mutates
``default_off``): off for unconfigured platforms as a security opt-in, but a saved list is that opt-in."""
if explicitly_configured:
default_off -= {ts for ts in default_off if platform in (_TOOLSET_PLATFORM_RESTRICTIONS.get(ts) or ())}
#: Toolsets young enough that absence from a saved ``platform_toolsets`` list means "never offered", not
#: "declined": saving ``hermes tools`` freezes a platform's composite into an explicit list nothing adds to, so
#: a later toolset stays off forever for picker users while ``[hermes-cli]`` users inherit it.
@@ -500,13 +493,16 @@ def _configurable_subset_of(tool_names: Set[str], platform: str) -> Set[str]:
def _default_off_toolsets(platform: str, explicitly_configured: bool) -> Set[str]:
"""Toolsets to strip from an implicit (composite-derived) enable set. A platform named after a default-off
toolset (``homeassistant``) keeps it, except platform-restricted ones (``discord`` on discord stays OFF); a
configured HASS_TOKEN is an explicit opt-in that must survive platforms resolving without a saved list."""
configured HASS_TOKEN is an explicit opt-in that must survive platforms resolving without a saved list.
Platform-native default-off toolsets (``discord`` on discord) are off for unconfigured platforms as a
security opt-in — an explicitly saved list IS that opt-in and lets them through."""
default_off = set(_DEFAULT_OFF_TOOLSETS)
if platform in default_off and platform not in _TOOLSET_PLATFORM_RESTRICTIONS:
default_off.remove(platform)
if "homeassistant" in default_off and _homeassistant_credentials_present():
default_off.remove("homeassistant")
_exempt_explicit_platform_native(default_off, platform, explicitly_configured=explicitly_configured)
if explicitly_configured:
default_off -= {ts for ts in default_off if platform in (_TOOLSET_PLATFORM_RESTRICTIONS.get(ts) or ())}
return default_off
@@ -571,7 +567,7 @@ def _get_platform_tools(config: dict, platform: str, *, include_default_mcp_serv
platform_toolsets = config.get("platform_toolsets") or {}
toolset_names = platform_toolsets.get(platform)
# An explicitly saved list (even a composite like ``hermes-discord``) is an opt-in to the platform's
# native default-off toolsets — see _exempt_explicit_platform_native.
# native default-off toolsets — see _default_off_toolsets.
explicitly_configured = isinstance(toolset_names, list)
if not explicitly_configured:
toolset_names = [_platform_default_toolset(platform)]
@@ -988,40 +984,30 @@ def _apply_platform_checklist(config: dict, pkey: str, new_enabled: Set[str], pr
_save_platform_tools(config, pkey, new_enabled)
def _configure_all_platforms(config: dict, platform_keys: List[str]) -> bool:
"""'Configure all platforms (global)' menu entry. Returns True when config was saved."""
all_current: Set[str] = set()
for pk in platform_keys:
all_current |= _current_platform_tools(config, pk)
new_enabled = _prompt_toolset_checklist("All platforms", all_current, force_fresh=True)
def _configure_platforms(config: dict, platform_keys: List[str], *, all_platforms: bool = False) -> bool:
"""Checklist + key setup + save for one platform, or for every platform at once (the 'Configure all
platforms (global)' menu entry). Returns True when config was saved."""
label = "All platforms" if all_platforms else PLATFORMS[platform_keys[0]]["label"]
current = {pk: _current_platform_tools(config, pk) for pk in platform_keys}
all_current = set().union(*current.values())
new_enabled = _prompt_toolset_checklist(label, all_current, force_fresh=True)
selected_to_configure = _toolsets_needing_setup(new_enabled, config)
_configure_list(selected_to_configure, config)
if new_enabled == all_current and not selected_to_configure:
print(color(" No changes", Colors.DIM))
print(color(" No changes" if all_platforms else f" No changes to {label}", Colors.DIM))
return False
for pk in platform_keys:
_apply_platform_checklist(config, pk, new_enabled, _current_platform_tools(config, pk),
set(selected_to_configure), indent=" ", header=True)
# Global: re-read after each save — reconciling agent.disabled_toolsets for one platform can change
# what the next platform resolves to. Single platform: diff against the pre-checklist snapshot.
prev = _current_platform_tools(config, pk) if all_platforms else current[pk]
_apply_platform_checklist(config, pk, new_enabled, prev, set(selected_to_configure),
indent=" " if all_platforms else " ", header=all_platforms)
save_config(config)
print(color(" ✓ Saved configuration for all platforms", Colors.GREEN))
print(color(" ✓ Saved configuration for all platforms" if all_platforms else f" ✓ Saved {label} configuration",
Colors.GREEN))
return True
def _configure_one_platform(config: dict, pkey: str) -> None:
"""Per-platform checklist + key setup + save."""
pinfo = PLATFORMS[pkey]
current_enabled = _current_platform_tools(config, pkey)
new_enabled = _prompt_toolset_checklist(pinfo["label"], current_enabled, force_fresh=True)
selected_to_configure = _toolsets_needing_setup(new_enabled, config)
_configure_list(selected_to_configure, config)
if new_enabled == current_enabled and not selected_to_configure:
print(color(f" No changes to {pinfo['label']}", Colors.DIM))
return
_apply_platform_checklist(config, pkey, new_enabled, current_enabled, set(selected_to_configure))
save_config(config)
print(color(f" ✓ Saved {pinfo['label']} configuration", Colors.GREEN))
def tools_command(args=None, first_install: bool = False, config: dict = None):
"""Entry point for `hermes tools` / `hermes setup tools`. ``first_install`` skips the menu (checklist + key
prompts); a wizard-passed ``config`` receives platform_toolsets so its final save_config() keeps them."""
@@ -1072,11 +1058,11 @@ def tools_command(args=None, first_install: bool = False, config: dict = None):
elif idx == mcp_idx:
_configure_mcp_tools_interactive(config)
elif idx == global_idx:
if _configure_all_platforms(config, platform_keys):
if _configure_platforms(config, platform_keys, all_platforms=True):
for ci, pk in enumerate(platform_keys):
platform_choices[ci] = _platform_menu_label(config, pk)
else:
_configure_one_platform(config, platform_keys[idx])
_configure_platforms(config, [platform_keys[idx]])
platform_choices[idx] = _platform_menu_label(config, platform_keys[idx])
print()

192
hermes_cli/tools_config_providers.py
View File

@@ -98,11 +98,9 @@ def _plugin_rows_for(category: str) -> list[dict]:
return _plugin_provider_rows(module, markers, **kwargs)
_plugin_image_gen_providers = partial(_plugin_rows_for, "image_gen")
_plugin_video_gen_providers = partial(_plugin_rows_for, "video_gen") # the only Video Generation rows
_plugin_web_search_providers = partial(_plugin_rows_for, "web")
_plugin_browser_providers = partial(_plugin_rows_for, "browser")
_plugin_tts_providers = partial(_plugin_rows_for, "tts")
_plugin_image_gen_providers, _plugin_video_gen_providers, _plugin_web_search_providers, \
_plugin_browser_providers, _plugin_tts_providers = (
partial(_plugin_rows_for, cat) for cat in ("image_gen", "video_gen", "web", "browser", "tts"))
def web_provider_capabilities(backend: str) -> list:
@@ -238,8 +236,6 @@ def _toolset_needs_configuration_prompt(ts_key: str, config: dict, *, force_fres
def _any_plugin_provider_available(registry_module: str) -> bool:
"""True when any provider in the plugin registry reports ``is_available()``."""
registry = _plugin_registry(registry_module)
if registry is None:
return False
try:
for provider in registry.list_providers():
try:
@@ -258,27 +254,21 @@ def _configure_tool_category(ts_key: str, cat: dict, config: dict, *, force_fres
chosen provider goes through the key-update prompts instead of the new-enable prompts."""
from hermes_cli.tools_config import _prompt_choice, _provider_env_ready, get_nous_subscription_features
icon = cat.get("icon", "")
name = cat["name"]
providers = _visible_providers(cat, config, force_fresh=force_fresh)
if len(providers) == 1:
provider = providers[0]
print()
print(color(f" --- {icon} {name} ({provider['name']}) ---", Colors.CYAN))
if not reconfigure:
if provider.get("tag"):
_print_info(f" {provider['tag']}")
if cat.get("setup_note"):
_print_info(f" {cat['setup_note']}")
_configure_provider(provider, config, force_fresh=force_fresh, reconfigure=reconfigure)
return
single = len(providers) == 1
title = "Choose a provider" if reconfigure else cat.get("setup_title", "Choose a provider")
print()
title = "Choose a provider" if reconfigure else cat.get("setup_title", "Choose a provider")
print(color(f" --- {icon} {name} - {title} ---", Colors.CYAN))
heading = f"({providers[0]['name']})" if single else f"- {title}"
print(color(f" --- {cat.get('icon', '')} {name} {heading} ---", Colors.CYAN))
if single and not reconfigure and providers[0].get("tag"):
_print_info(f" {providers[0]['tag']}")
if not reconfigure and cat.get("setup_note"):
_print_info(f" {cat['setup_note']}")
if single:
_configure_provider(providers[0], config, force_fresh=force_fresh, reconfigure=reconfigure)
return
print()
# Logged-in Nous users get a marker on rows included in their subscription (cost-extra vs. included).
@@ -327,9 +317,7 @@ def _web_tier_matches(provider: dict, config: dict) -> bool:
if not row_tier:
return True
web_cfg = config.get("web") if isinstance(config.get("web"), dict) else {}
tiers = web_cfg.get("provider_tier")
if not isinstance(tiers, dict):
tiers = {}
tiers = web_cfg.get("provider_tier") if isinstance(web_cfg.get("provider_tier"), dict) else {}
configured = str(tiers.get(provider["web_backend"], "") or "").lower().strip()
if configured in ("free", "paid"):
return configured == row_tier
@@ -459,20 +447,16 @@ _ACTIVE_CHECKS: tuple[tuple[str, Callable[[dict, dict], bool]], ...] = (
def _is_provider_active(provider: dict, config: dict, *, force_fresh: bool = False) -> bool:
"""Check if a provider entry matches the currently active config."""
managed_feature = provider.get("managed_nous_feature")
plugin_name = provider.get("image_gen_plugin_name")
if plugin_name and not managed_feature:
# Managed entries fall through to the managed branch, which also checks use_gateway — otherwise a
# managed FAL pick and a direct-key FAL pick would both report active.
image_cfg = config.get("image_gen", {})
if not (isinstance(image_cfg, dict) and image_cfg.get("provider") == plugin_name):
return False
# A direct-key entry is only active when the managed route is OFF.
return not is_truthy_value(image_cfg.get("use_gateway"), default=False)
video_plugin_name = provider.get("video_gen_plugin_name")
if video_plugin_name and not managed_feature:
video_cfg = config.get("video_gen", {})
return isinstance(video_cfg, dict) and video_cfg.get("provider") == video_plugin_name
# Managed entries fall through to the managed branch, which also checks use_gateway — otherwise a
# managed FAL pick and a direct-key FAL pick would both report active.
for section in ("image_gen", "video_gen"):
plugin_name = provider.get(f"{section}_plugin_name")
if plugin_name and not managed_feature:
gen_cfg = config.get(section, {})
if not (isinstance(gen_cfg, dict) and gen_cfg.get("provider") == plugin_name):
return False
# A direct-key image gen entry is only active when the managed route is OFF.
return section == "video_gen" or not is_truthy_value(gen_cfg.get("use_gateway"), default=False)
if managed_feature:
return _managed_provider_active(provider, config, managed_feature, force_fresh)
@@ -511,21 +495,13 @@ def _plugin_model_catalog(registry_module: str, plugin_name: str):
"""``(catalog_dict, default_model_id)`` for a plugin provider; ``catalog_dict`` is shaped like the legacy
``FAL_MODELS`` table so the picker path is shared. ``({}, None)`` if unregistered or no models."""
registry = _plugin_registry(registry_module)
if registry is None:
return {}, None
try:
try: # a missing registry / unknown provider surfaces as AttributeError here — same ({}, None) outcome
provider = registry.get_provider(plugin_name)
except Exception:
return {}, None
if provider is None:
return {}, None
try:
models = provider.list_models() or []
default = provider.default_model()
except Exception:
return {}, None
catalog = {m["id"]: m for m in models if isinstance(m, dict) and "id" in m}
return catalog, default
return {m["id"]: m for m in models if isinstance(m, dict) and "id" in m}, default
_plugin_image_gen_catalog = partial(_plugin_model_catalog, "agent.image_gen_registry")
@@ -585,16 +561,17 @@ def _configure_imagegen_model(backend_name: str, config: dict) -> None:
_pick_model_from_catalog(catalog, default_model, backend["config_key"], backend["display"], config)
def _configure_imagegen_model_for_plugin(plugin_name: str, config: dict) -> None:
"""Prompt for a model from a plugin-registered image gen backend's catalog."""
catalog, default_model = _plugin_image_gen_catalog(plugin_name)
_pick_model_from_catalog(catalog, default_model, "image_gen", plugin_name, config)
def _configure_gen_model_for_plugin(section: str, plugin_name: str, config: dict) -> None:
"""Prompt for a model from a plugin-registered image/video gen catalog (video rows keep their historical
two-space indent)."""
catalog_fn = _plugin_image_gen_catalog if section == "image_gen" else _plugin_video_gen_catalog
catalog, default_model = catalog_fn(plugin_name)
_pick_model_from_catalog(catalog, default_model, section, plugin_name, config,
row_indent="" if section == "image_gen" else " ")
def _configure_videogen_model_for_plugin(plugin_name: str, config: dict) -> None:
"""Prompt for a video gen model from a plugin's catalog (rows keep their historical two-space indent)."""
catalog, default_model = _plugin_video_gen_catalog(plugin_name)
_pick_model_from_catalog(catalog, default_model, "video_gen", plugin_name, config, row_indent=" ")
_configure_imagegen_model_for_plugin = partial(_configure_gen_model_for_plugin, "image_gen")
_configure_videogen_model_for_plugin = partial(_configure_gen_model_for_plugin, "video_gen")
def _configure_xai_imagine_storage(section_name: str, config: dict) -> None:
@@ -619,21 +596,23 @@ def _configure_xai_imagine_storage(section_name: str, config: dict) -> None:
else " xAI stored public URLs enabled without automatic expiry")
def _select_plugin_gen_provider(section: str, plugin_name: str, config: dict, *, use_gateway: bool = False) -> None:
"""Persist a plugin-backed image/video gen provider selection and run its model picker.
``use_gateway=True`` (Nous-managed pick) stores ``<section>.provider: nous``; BYOK picks store the
plugin name. Any legacy ``use_gateway`` key is removed so old read-time shims cannot override the
selection."""
def _select_into(config: dict, section: str, key: str, vendor, managed) -> dict:
"""Write ``config[section][key] = vendor`` (``nous`` for a managed pick) and drop any legacy ``use_gateway``
key so the old read-time shim cannot override the new choice. Returns the section dict."""
from hermes_cli.tools_config import _cfg_section
cfg = _cfg_section(config, section)
cfg["provider"] = NOUS_MANAGED_PROVIDER if use_gateway else plugin_name
cfg[key] = NOUS_MANAGED_PROVIDER if managed else vendor
cfg.pop("use_gateway", None)
return cfg
def _select_plugin_gen_provider(section: str, plugin_name: str, config: dict, *, use_gateway: bool = False) -> None:
"""Persist a plugin-backed image/video gen provider selection (``nous`` for a Nous-managed pick, else the
plugin name) and run its model picker."""
cfg = _select_into(config, section, "provider", plugin_name, use_gateway)
_print_success(f" {section}.provider set to: {cfg['provider']}")
if section == "image_gen":
_configure_imagegen_model_for_plugin(plugin_name, config)
else:
_configure_videogen_model_for_plugin(plugin_name, config)
_configure_gen_model_for_plugin(section, plugin_name, config)
if plugin_name == "xai":
_configure_xai_imagine_storage(section, config)
@@ -678,37 +657,25 @@ _PROVIDER_MARKER_SECTIONS = {
}
def _drop_use_gateway(section) -> None:
if isinstance(section, dict):
section.pop("use_gateway", None)
def _write_provider_config(provider: dict, config: dict, *, managed_feature) -> None:
"""Persist the provider/backend config keys for a selected provider.
Pure, non-interactive core of :func:`_configure_provider` (no env prompts, post-setup hooks, Nous
auth gating or model pickers) shared by the CLI and the GUI ``PUT .../provider`` endpoint. Each pick
writes exactly ONE provider string per category (``nous`` for managed rows) and removes any legacy
``use_gateway`` key so the read-time shim cannot override the new choice."""
from hermes_cli.tools_config import TOOL_CATEGORIES, _cfg_section
from hermes_cli.tools_config import TOOL_CATEGORIES
def _set_selection(section_key: str, name_key: str, vendor_value) -> None:
section = _cfg_section(config, section_key)
section[name_key] = NOUS_MANAGED_PROVIDER if managed_feature else vendor_value
_drop_use_gateway(section)
if provider.get("tts_provider"):
_set_selection("tts", "provider", provider["tts_provider"])
if provider.get("stt_provider"):
_set_selection("stt", "provider", provider["stt_provider"])
for marker, section_key in (("tts_provider", "tts"), ("stt_provider", "stt")):
if provider.get(marker):
_select_into(config, section_key, "provider", provider[marker], managed_feature)
if "browser_provider" in provider:
bp = provider["browser_provider"]
browser_cfg = config.setdefault("browser", {})
if bp or managed_feature:
# Browser Use mode (browser.backend) composes with the provider — keep the driver choice intact.
_set_selection("browser", "cloud_provider", bp)
_select_into(config, "browser", "cloud_provider", bp, managed_feature)
else:
browser_cfg.pop("use_gateway", None)
config.setdefault("browser", {}).pop("use_gateway", None)
if provider.get("browser_backend"):
config.setdefault("browser", {})["backend"] = provider["browser_backend"]
# Local engine rows ("Local Browser" resets to auto, "Lightpanda" sets lightpanda); composes with browser.backend.
@@ -716,16 +683,14 @@ def _write_provider_config(provider: dict, config: dict, *, managed_feature) ->
config.setdefault("browser", {})["engine"] = provider["browser_engine"]
if provider.get("web_backend"):
_set_selection("web", "backend", provider["web_backend"])
web_cfg = config.get("web")
if isinstance(web_cfg, dict):
tier = provider.get("web_tier")
tiers = web_cfg.setdefault("provider_tier", {}) if tier else web_cfg.get("provider_tier")
if isinstance(tiers, dict):
if tier:
tiers[provider["web_backend"]] = tier
else:
tiers.pop(provider["web_backend"], None)
web_cfg = _select_into(config, "web", "backend", provider["web_backend"], managed_feature)
tier = provider.get("web_tier")
tiers = web_cfg.setdefault("provider_tier", {}) if tier else web_cfg.get("provider_tier")
if isinstance(tiers, dict):
if tier:
tiers[provider["web_backend"]] = tier
else:
tiers.pop(provider["web_backend"], None)
if provider.get("computer_use_backend"):
config.setdefault("computer_use", {})["backend"] = provider["computer_use_backend"]
@@ -736,7 +701,7 @@ def _write_provider_config(provider: dict, config: dict, *, managed_feature) ->
section = config.setdefault(managed_feature, {})
if isinstance(section, dict):
section["provider"] = NOUS_MANAGED_PROVIDER
_drop_use_gateway(section)
section.pop("use_gateway", None)
elif not managed_feature:
# Non-gateway pick — clear any stale legacy use_gateway key on the category. Resolve the category from
# the row's own markers first (plugin-injected rows are NOT in TOOL_CATEGORIES' hardcoded lists), then
@@ -745,14 +710,15 @@ def _write_provider_config(provider: dict, config: dict, *, managed_feature) ->
if not sections:
sections = [cat_key for cat_key, cat in TOOL_CATEGORIES.items() if provider in cat.get("providers", [])][:1]
for section_key in sections:
_drop_use_gateway(config.get(section_key))
if isinstance(config.get(section_key), dict):
config[section_key].pop("use_gateway", None)
def apply_provider_selection(ts_key: str, provider_name: str, config: dict) -> None:
"""Non-interactively persist a provider selection for a toolset (config keys only — API keys, post-setup
hooks, auth gating and model pickers are separate GUI endpoints). ``provider_name`` is resolved among
:func:`_visible_providers` rows; raises ``KeyError`` for an unknown toolset or provider."""
from hermes_cli.tools_config import TOOL_CATEGORIES, _cfg_section
from hermes_cli.tools_config import TOOL_CATEGORIES
cat = TOOL_CATEGORIES.get(ts_key)
if cat is None:
@@ -775,9 +741,7 @@ def apply_provider_selection(ts_key: str, provider_name: str, config: dict) -> N
("image_gen", "fal" if provider.get("imagegen_backend") and not managed_feature else None)]
for section_key, vendor in selections:
if vendor:
section = _cfg_section(config, section_key)
section["provider"] = NOUS_MANAGED_PROVIDER if managed_feature else vendor
section.pop("use_gateway", None)
_select_into(config, section_key, "provider", vendor, managed_feature)
def _nous_provider_gate(provider: dict, config: dict, managed_feature, *, force_fresh: bool) -> bool:
@@ -811,24 +775,17 @@ def _nous_provider_gate(provider: dict, config: dict, managed_feature, *, force_
def _finish_provider_selection(provider: dict, config: dict, managed_feature) -> None:
"""Model pickers that follow a provider pick: plugin image/video gen, in-tree FAL, STT."""
from hermes_cli.tools_config import _cfg_section
plugin_name = provider.get("image_gen_plugin_name")
if plugin_name:
_select_plugin_image_gen_provider(plugin_name, config, use_gateway=bool(managed_feature))
return
video_plugin = provider.get("video_gen_plugin_name")
if video_plugin:
_select_plugin_video_gen_provider(video_plugin, config, use_gateway=bool(managed_feature))
return
for section in ("image_gen", "video_gen"):
plugin_name = provider.get(f"{section}_plugin_name")
if plugin_name:
_select_plugin_gen_provider(section, plugin_name, config, use_gateway=bool(managed_feature))
return
backend = provider.get("imagegen_backend")
if backend:
_configure_imagegen_model(backend, config)
# In-tree FAL is the only non-plugin backend: "nous" for a managed row, "fal" for BYOK, drop legacy
# use_gateway — never clobber a managed pick back onto direct keys.
img_cfg = _cfg_section(config, "image_gen")
img_cfg["provider"] = NOUS_MANAGED_PROVIDER if managed_feature else "fal"
img_cfg.pop("use_gateway", None)
_select_into(config, "image_gen", "provider", "fal", managed_feature)
# STT rows prompt for a model after the pick (skipped for managed rows — the gateway pins it).
if provider.get("stt_provider") and not managed_feature:
_configure_stt_model(provider["stt_provider"], config)
@@ -883,10 +840,9 @@ def _prompt_secret(
Reconfigure mode shows the current value and re-prompts ("Enter to keep current"); the new-enable
flow prompts with ``default_val`` visible when one exists, else as a password, and ``strip`` decides
whether whitespace is trimmed (and whitespace-only counts as skipped)."""
if reconfigure:
existing = get_env_value(key)
if existing:
_print_info(f" {key}: configured ({existing[:8]}...)")
existing = get_env_value(key) if reconfigure else ""
if existing:
_print_info(f" {key}: configured ({existing[:8]}...)")
if url:
_print_info(f" {url_label}: {url}")
if reconfigure: