diff --git a/hermes_cli/mcp_config.py b/hermes_cli/mcp_config.py index ae845b069b..0e01f558dd 100644 --- a/hermes_cli/mcp_config.py +++ b/hermes_cli/mcp_config.py @@ -16,6 +16,7 @@ import time from typing import Any, Dict, List, Optional, Tuple from hermes_cli.config import ( + cfg_get, load_config, save_config, get_env_value, @@ -716,7 +717,7 @@ def cmd_mcp_configure(args): # Update config config = load_config() - server_entry = config.get("mcp_servers", {}).get(name, {}) + server_entry = cfg_get(config, "mcp_servers", name, default={}) if len(chosen) == total: # All selected → remove include/exclude (register all) diff --git a/hermes_cli/plugins.py b/hermes_cli/plugins.py index 3d514ddc12..62e9ac583c 100644 --- a/hermes_cli/plugins.py +++ b/hermes_cli/plugins.py @@ -45,6 +45,7 @@ from typing import Any, Callable, Dict, List, Optional, Set, Union from hermes_constants import get_hermes_home from utils import env_var_enabled +from hermes_cli.config import cfg_get try: import yaml @@ -115,7 +116,7 @@ def _get_disabled_plugins() -> set: try: from hermes_cli.config import load_config config = load_config() - disabled = config.get("plugins", {}).get("disabled", []) + disabled = cfg_get(config, "plugins", "disabled", default=[]) return set(disabled) if isinstance(disabled, list) else set() except Exception: return set() diff --git a/hermes_cli/plugins_cmd.py b/hermes_cli/plugins_cmd.py index 349a11de11..a2d1730993 100644 --- a/hermes_cli/plugins_cmd.py +++ b/hermes_cli/plugins_cmd.py @@ -18,6 +18,7 @@ from pathlib import Path from typing import Optional from hermes_constants import get_hermes_home +from hermes_cli.config import cfg_get logger = logging.getLogger(__name__) @@ -519,7 +520,7 @@ def _get_disabled_set() -> set: try: from hermes_cli.config import load_config config = load_config() - disabled = config.get("plugins", {}).get("disabled", []) + disabled = cfg_get(config, "plugins", "disabled", default=[]) return set(disabled) if isinstance(disabled, list) else set() except Exception: return set() @@ -763,7 +764,7 @@ def _get_current_memory_provider() -> str: try: from hermes_cli.config import load_config config = load_config() - return config.get("memory", {}).get("provider", "") or "" + return cfg_get(config, "memory", "provider", default="") or "" except Exception: return "" @@ -773,7 +774,7 @@ def _get_current_context_engine() -> str: try: from hermes_cli.config import load_config config = load_config() - return config.get("context", {}).get("engine", "compressor") or "compressor" + return cfg_get(config, "context", "engine", default="compressor") or "compressor" except Exception: return "compressor" diff --git a/hermes_cli/setup.py b/hermes_cli/setup.py index 011b4575e4..921c06f74e 100644 --- a/hermes_cli/setup.py +++ b/hermes_cli/setup.py @@ -131,6 +131,7 @@ def _set_reasoning_effort(config: Dict[str, Any], effort: str) -> None: # Import config helpers from hermes_cli.config import ( + cfg_get, DEFAULT_CONFIG, get_hermes_home, get_config_path, @@ -441,7 +442,7 @@ def _print_setup_summary(config: dict, hermes_home): tool_status.append(("Image Generation", False, "FAL_KEY or OPENAI_API_KEY")) # TTS — show configured provider - tts_provider = config.get("tts", {}).get("provider", "edge") + tts_provider = cfg_get(config, "tts", "provider", default="edge") if subscription_features.tts.managed_by_nous: tool_status.append(("Text-to-Speech (OpenAI via Nous subscription)", True, None)) elif tts_provider == "elevenlabs" and get_env_value("ELEVENLABS_API_KEY"): @@ -480,7 +481,7 @@ def _print_setup_summary(config: dict, hermes_home): if subscription_features.modal.managed_by_nous: tool_status.append(("Modal Execution (Nous subscription)", True, None)) - elif config.get("terminal", {}).get("backend") == "modal": + elif cfg_get(config, "terminal", "backend") == "modal": if subscription_features.modal.direct_override: tool_status.append(("Modal Execution (direct Modal)", True, None)) else: @@ -1179,7 +1180,7 @@ def setup_terminal_backend(config: dict): print_info(f" Guide: {_DOCS_BASE}/developer-guide/environments") print() - current_backend = config.get("terminal", {}).get("backend", "local") + current_backend = cfg_get(config, "terminal", "backend", default="local") is_linux = _platform.system() == "Linux" # Build backend choices with descriptions @@ -1228,7 +1229,7 @@ def setup_terminal_backend(config: dict): print_info( " the agent starts. CLI mode always starts in the current directory." ) - current_cwd = config.get("terminal", {}).get("cwd", "") + current_cwd = cfg_get(config, "terminal", "cwd", default="") cwd = prompt(" Messaging working directory", current_cwd or str(Path.home())) if cwd: config["terminal"]["cwd"] = cwd @@ -1259,9 +1260,7 @@ def setup_terminal_backend(config: dict): print_info(f"Docker found: {docker_bin}") # Docker image - current_image = config.get("terminal", {}).get( - "docker_image", "nikolaik/python-nodejs:python3.11-nodejs20" - ) + current_image = cfg_get(config, "terminal", "docker_image", default="nikolaik/python-nodejs:python3.11-nodejs20") image = prompt(" Docker image", current_image) config["terminal"]["docker_image"] = image save_env_value("TERMINAL_DOCKER_IMAGE", image) @@ -1281,9 +1280,7 @@ def setup_terminal_backend(config: dict): else: print_info(f"Found: {sing_bin}") - current_image = config.get("terminal", {}).get( - "singularity_image", "docker://nikolaik/python-nodejs:python3.11-nodejs20" - ) + current_image = cfg_get(config, "terminal", "singularity_image", default="docker://nikolaik/python-nodejs:python3.11-nodejs20") image = prompt(" Container image", current_image) config["terminal"]["singularity_image"] = image save_env_value("TERMINAL_SINGULARITY_IMAGE", image) @@ -1302,7 +1299,7 @@ def setup_terminal_backend(config: dict): get_nous_subscription_features(config).nous_auth_present and is_managed_tool_gateway_ready("modal") ) - modal_mode = normalize_modal_mode(config.get("terminal", {}).get("modal_mode")) + modal_mode = normalize_modal_mode(cfg_get(config, "terminal", "modal_mode")) use_managed_modal = False if managed_modal_available: modal_choices = [ @@ -1439,9 +1436,7 @@ def setup_terminal_backend(config: dict): print_success(" Configured") # Daytona image - current_image = config.get("terminal", {}).get( - "daytona_image", "nikolaik/python-nodejs:python3.11-nodejs20" - ) + current_image = cfg_get(config, "terminal", "daytona_image", default="nikolaik/python-nodejs:python3.11-nodejs20") image = prompt(" Sandbox image", current_image) config["terminal"]["daytona_image"] = image save_env_value("TERMINAL_DAYTONA_IMAGE", image) @@ -1545,7 +1540,7 @@ def setup_agent_settings(config: dict): # ── Max Iterations ── current_max = get_env_value("HERMES_MAX_ITERATIONS") or str( - config.get("agent", {}).get("max_turns", 90) + cfg_get(config, "agent", "max_turns", default=90) ) print_info("Maximum tool-calling iterations per conversation.") print_info("Higher = more complex tasks, but costs more tokens.") @@ -1573,7 +1568,7 @@ def setup_agent_settings(config: dict): print_info(" all — Show every tool call with a short preview") print_info(" verbose — Full args, results, and debug logs") - current_mode = config.get("display", {}).get("tool_progress", "all") + current_mode = cfg_get(config, "display", "tool_progress", default="all") mode = prompt("Tool progress mode", current_mode) if mode.lower() in ("off", "new", "all", "verbose"): if "display" not in config: @@ -1593,7 +1588,7 @@ def setup_agent_settings(config: dict): config.setdefault("compression", {})["enabled"] = True - current_threshold = config.get("compression", {}).get("threshold", 0.50) + current_threshold = cfg_get(config, "compression", "threshold", default=0.50) threshold_str = prompt("Compression threshold (0.5-0.95)", str(current_threshold)) try: threshold = float(threshold_str) @@ -2601,11 +2596,11 @@ def _get_section_config_summary(config: dict, section_key: str) -> Optional[str] return "configured" elif section_key == "terminal": - backend = config.get("terminal", {}).get("backend", "local") + backend = cfg_get(config, "terminal", "backend", default="local") return f"backend: {backend}" elif section_key == "agent": - max_turns = config.get("agent", {}).get("max_turns", 90) + max_turns = cfg_get(config, "agent", "max_turns", default=90) return f"max turns: {max_turns}" elif section_key == "gateway": diff --git a/hermes_cli/skills_config.py b/hermes_cli/skills_config.py index 741a8b8341..8eaf64605a 100644 --- a/hermes_cli/skills_config.py +++ b/hermes_cli/skills_config.py @@ -13,7 +13,7 @@ Config stored in ~/.hermes/config.yaml under: """ from typing import List, Optional, Set -from hermes_cli.config import load_config, save_config +from hermes_cli.config import cfg_get, load_config, save_config from hermes_cli.colors import Colors, color from hermes_cli.platforms import PLATFORMS as _PLATFORMS @@ -30,7 +30,7 @@ def get_disabled_skills(config: dict, platform: Optional[str] = None) -> Set[str global_disabled = set(skills_cfg.get("disabled", [])) if platform is None: return global_disabled - platform_disabled = skills_cfg.get("platform_disabled", {}).get(platform) + platform_disabled = cfg_get(skills_cfg, "platform_disabled", platform) if platform_disabled is None: return global_disabled return set(platform_disabled) diff --git a/hermes_cli/status.py b/hermes_cli/status.py index 31aa1d5c2c..5e9b3dbcdf 100644 --- a/hermes_cli/status.py +++ b/hermes_cli/status.py @@ -13,7 +13,7 @@ PROJECT_ROOT = Path(__file__).parent.parent.resolve() from hermes_cli.auth import AuthError, resolve_provider from hermes_cli.colors import Colors, color -from hermes_cli.config import get_env_path, get_env_value, get_hermes_home, load_config +from hermes_cli.config import cfg_get, get_env_path, get_env_value, get_hermes_home, load_config from hermes_cli.models import provider_label from hermes_cli.nous_subscription import get_nous_subscription_features from hermes_cli.runtime_provider import resolve_requested_provider @@ -306,7 +306,7 @@ def show_status(args): # (hermes status doesn't go through cli.py's config loading) try: _cfg = load_config() - terminal_env = _cfg.get("terminal", {}).get("backend", "local") + terminal_env = cfg_get(_cfg, "terminal", "backend", default="local") except Exception: terminal_env = "local" print(f" Backend: {terminal_env}") diff --git a/hermes_cli/tools_config.py b/hermes_cli/tools_config.py index b4a19f0bc4..31156f15f9 100644 --- a/hermes_cli/tools_config.py +++ b/hermes_cli/tools_config.py @@ -18,6 +18,7 @@ from typing import Dict, List, Optional, Set from hermes_cli.config import ( + cfg_get, load_config, save_config, get_env_value, save_env_value, ) from hermes_cli.colors import Colors, color @@ -965,7 +966,7 @@ def _save_platform_tools(config: dict, platform: str, enabled_toolset_keys: Set[ platform_default_keys = {p["default_toolset"] for p in PLATFORMS.values()} # Get existing toolsets for this platform - existing_toolsets = config.get("platform_toolsets", {}).get(platform, []) + existing_toolsets = cfg_get(config, "platform_toolsets", platform, default=[]) if not isinstance(existing_toolsets, list): existing_toolsets = [] existing_toolsets = [str(ts) for ts in existing_toolsets] @@ -1352,23 +1353,23 @@ def _is_provider_active(provider: dict, config: dict) -> bool: if provider.get("tts_provider"): return ( feature.managed_by_nous - and config.get("tts", {}).get("provider") == provider["tts_provider"] + and cfg_get(config, "tts", "provider") == provider["tts_provider"] ) if "browser_provider" in provider: - current = config.get("browser", {}).get("cloud_provider") + current = cfg_get(config, "browser", "cloud_provider") return feature.managed_by_nous and provider["browser_provider"] == current if provider.get("web_backend"): - current = config.get("web", {}).get("backend") + current = cfg_get(config, "web", "backend") return feature.managed_by_nous and current == provider["web_backend"] return feature.managed_by_nous if provider.get("tts_provider"): - return config.get("tts", {}).get("provider") == provider["tts_provider"] + return cfg_get(config, "tts", "provider") == provider["tts_provider"] if "browser_provider" in provider: - current = config.get("browser", {}).get("cloud_provider") + current = cfg_get(config, "browser", "cloud_provider") return provider["browser_provider"] == current if provider.get("web_backend"): - current = config.get("web", {}).get("backend") + current = cfg_get(config, "web", "backend") return current == provider["web_backend"] if provider.get("imagegen_backend"): image_cfg = config.get("image_gen", {}) diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index b91edc16d1..40f65e5e76 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -33,6 +33,7 @@ if str(PROJECT_ROOT) not in sys.path: from hermes_cli import __version__, __release_date__ from hermes_cli.config import ( + cfg_get, DEFAULT_CONFIG, OPTIONAL_ENV_VARS, get_config_path, @@ -2902,7 +2903,7 @@ async def get_dashboard_themes(): them without a stub. """ config = load_config() - active = config.get("dashboard", {}).get("theme", "default") + active = cfg_get(config, "dashboard", "theme", default="default") user_themes = _discover_user_themes() seen = set() themes = [] diff --git a/hermes_cli/webhook.py b/hermes_cli/webhook.py index 0ec4c6784b..4b74204bcc 100644 --- a/hermes_cli/webhook.py +++ b/hermes_cli/webhook.py @@ -19,6 +19,7 @@ from typing import Dict from hermes_constants import display_hermes_home from utils import atomic_replace +from hermes_cli.config import cfg_get _SUBSCRIPTIONS_FILENAME = "webhook_subscriptions.json" @@ -60,7 +61,7 @@ def _get_webhook_config() -> dict: try: from hermes_cli.config import load_config cfg = load_config() - return cfg.get("platforms", {}).get("webhook", {}) + return cfg_get(cfg, "platforms", "webhook", default={}) except Exception: return {} diff --git a/plugins/memory/__init__.py b/plugins/memory/__init__.py index 0ae65a25d5..0d714f64dd 100644 --- a/plugins/memory/__init__.py +++ b/plugins/memory/__init__.py @@ -27,6 +27,7 @@ import logging import sys from pathlib import Path from typing import List, Optional, Tuple +from hermes_cli.config import cfg_get logger = logging.getLogger(__name__) @@ -314,7 +315,7 @@ def _get_active_memory_provider() -> Optional[str]: try: from hermes_cli.config import load_config config = load_config() - return config.get("memory", {}).get("provider") or None + return cfg_get(config, "memory", "provider") or None except Exception: return None diff --git a/plugins/memory/hindsight/__init__.py b/plugins/memory/hindsight/__init__.py index 8bd45d8b3e..1710b74f6a 100644 --- a/plugins/memory/hindsight/__init__.py +++ b/plugins/memory/hindsight/__init__.py @@ -41,6 +41,7 @@ from typing import Any, Dict, List from agent.memory_provider import MemoryProvider from hermes_constants import get_hermes_home from tools.registry import tool_error +from hermes_cli.config import cfg_get logger = logging.getLogger(__name__) @@ -913,7 +914,7 @@ class HindsightMemoryProvider(MemoryProvider): self._api_url = self._config.get("api_url") or os.environ.get("HINDSIGHT_API_URL", default_url) self._llm_base_url = self._config.get("llm_base_url", "") - banks = self._config.get("banks", {}).get("hermes", {}) + banks = cfg_get(self._config, "banks", "hermes", default={}) static_bank_id = self._config.get("bank_id") or banks.get("bankId", "hermes") self._bank_id_template = self._config.get("bank_id_template", "") or "" self._bank_id = _resolve_bank_id_template( diff --git a/plugins/memory/holographic/__init__.py b/plugins/memory/holographic/__init__.py index cd4ef07b44..dc9ee530c5 100644 --- a/plugins/memory/holographic/__init__.py +++ b/plugins/memory/holographic/__init__.py @@ -26,6 +26,7 @@ from agent.memory_provider import MemoryProvider from tools.registry import tool_error from .store import MemoryStore from .retrieval import FactRetriever +from hermes_cli.config import cfg_get logger = logging.getLogger(__name__) @@ -102,7 +103,7 @@ def _load_plugin_config() -> dict: import yaml with open(config_path) as f: all_config = yaml.safe_load(f) or {} - return all_config.get("plugins", {}).get("hermes-memory-store", {}) or {} + return cfg_get(all_config, "plugins", "hermes-memory-store", default={}) or {} except Exception: return {} diff --git a/plugins/memory/honcho/cli.py b/plugins/memory/honcho/cli.py index 8f354d2cdb..402389ab96 100644 --- a/plugins/memory/honcho/cli.py +++ b/plugins/memory/honcho/cli.py @@ -12,6 +12,7 @@ from pathlib import Path from hermes_constants import get_hermes_home from plugins.memory.honcho.client import resolve_active_host, resolve_config_path, HOST +from hermes_cli.config import cfg_get def clone_honcho_for_profile(profile_name: str) -> bool: @@ -106,7 +107,7 @@ def cmd_enable(args) -> None: # If this is a new profile host block with no settings, clone from default if not block.get("aiPeer"): - default_block = cfg.get("hosts", {}).get(HOST, {}) + default_block = cfg_get(cfg, "hosts", HOST, default={}) for key in ("recallMode", "writeFrequency", "sessionStrategy", "contextTokens", "dialecticReasoningLevel", "dialecticDynamic", "dialecticMaxChars", "messageMaxChars", "dialecticMaxInputChars", @@ -139,7 +140,7 @@ def cmd_disable(args) -> None: cfg = _read_config() host = _host_key() label = f"[{host}] " if host != "hermes" else "" - block = cfg.get("hosts", {}).get(host, {}) + block = cfg_get(cfg, "hosts", host, default={}) if not block or block.get("enabled") is False: print(f" {label}Honcho is already disabled.\n") @@ -212,7 +213,7 @@ def sync_honcho_profiles_quiet() -> int: if not cfg: return 0 - default_block = cfg.get("hosts", {}).get(HOST, {}) + default_block = cfg_get(cfg, "hosts", HOST, default={}) has_key = bool(cfg.get("apiKey") or os.environ.get("HONCHO_API_KEY")) if not default_block and not has_key: return 0 diff --git a/run_agent.py b/run_agent.py index 24043a7012..7bb47fcdda 100644 --- a/run_agent.py +++ b/run_agent.py @@ -160,6 +160,7 @@ from agent.trajectory import ( save_trajectory as _save_trajectory_to_file, ) from utils import atomic_json_write, base_url_host_matches, base_url_hostname, env_var_enabled, normalize_proxy_url +from hermes_cli.config import cfg_get @@ -1788,7 +1789,7 @@ class AIAgent: # compression model. Custom endpoints often cannot report this via # /models, so the startup feasibility check needs the config hint. try: - _aux_cfg = _agent_cfg.get("auxiliary", {}).get("compression", {}) + _aux_cfg = cfg_get(_agent_cfg, "auxiliary", "compression", default={}) except Exception: _aux_cfg = {} if isinstance(_aux_cfg, dict):