diff --git a/cli.py b/cli.py index 0f2598f1b..943a1fa6f 100755 --- a/cli.py +++ b/cli.py @@ -3484,6 +3484,56 @@ class HermesCLI: except Exception as e: print(f" Error generating insights: {e}") + def _check_config_mcp_changes(self) -> None: + """Detect mcp_servers changes in config.yaml and auto-reload MCP connections. + + Called from process_loop every CONFIG_WATCH_INTERVAL seconds. + Compares config.yaml mtime + mcp_servers section against the last + known state. When a change is detected, triggers _reload_mcp() and + informs the user so they know the tool list has been refreshed. + """ + import time + import yaml as _yaml + + CONFIG_WATCH_INTERVAL = 5.0 # seconds between config.yaml stat() calls + + now = time.monotonic() + if now - self._last_config_check < CONFIG_WATCH_INTERVAL: + return + self._last_config_check = now + + from hermes_cli.config import get_config_path as _get_config_path + cfg_path = _get_config_path() + if not cfg_path.exists(): + return + + try: + mtime = cfg_path.stat().st_mtime + except OSError: + return + + if mtime == self._config_mtime: + return # File unchanged — fast path + + # File changed — check whether mcp_servers section changed + self._config_mtime = mtime + try: + with open(cfg_path, encoding="utf-8") as f: + new_cfg = _yaml.safe_load(f) or {} + except Exception: + return + + new_mcp = new_cfg.get("mcp_servers") or {} + if new_mcp == self._config_mcp_servers: + return # mcp_servers unchanged (some other section was edited) + + self._config_mcp_servers = new_mcp + # Notify user and reload + print() + print("🔄 MCP server config changed — reloading connections...") + with self._busy_command(self._slow_command_status("/reload-mcp")): + self._reload_mcp() + def _reload_mcp(self): """Reload MCP servers: disconnect all, re-read config.yaml, reconnect. @@ -4749,6 +4799,12 @@ class HermesCLI: self._interrupt_queue = queue.Queue() # For messages typed while agent is running self._should_exit = False self._last_ctrl_c_time = 0 # Track double Ctrl+C for force exit + # Config file watcher — detect mcp_servers changes and auto-reload + from hermes_cli.config import get_config_path as _get_config_path + _cfg_path = _get_config_path() + self._config_mtime: float = _cfg_path.stat().st_mtime if _cfg_path.exists() else 0.0 + self._config_mcp_servers: dict = self.config.get("mcp_servers") or {} + self._last_config_check: float = 0.0 # monotonic time of last check # Clarify tool state: interactive question/answer with the user. # When the agent calls the clarify tool, _clarify_state is set and @@ -5682,6 +5738,9 @@ class HermesCLI: try: user_input = self._pending_input.get(timeout=0.1) except queue.Empty: + # Periodic config watcher — auto-reload MCP on mcp_servers change + if not self._agent_running: + self._check_config_mcp_changes() continue if not user_input: diff --git a/tests/test_cli_mcp_config_watch.py b/tests/test_cli_mcp_config_watch.py new file mode 100644 index 000000000..067ecc4cf --- /dev/null +++ b/tests/test_cli_mcp_config_watch.py @@ -0,0 +1,103 @@ +"""Tests for automatic MCP reload when config.yaml mcp_servers section changes.""" +import time +from pathlib import Path +from unittest.mock import MagicMock, patch + + +def _make_cli(tmp_path, mcp_servers=None): + """Create a minimal HermesCLI instance with mocked config.""" + import cli as cli_mod + obj = object.__new__(cli_mod.HermesCLI) + obj.config = {"mcp_servers": mcp_servers or {}} + obj._agent_running = False + obj._last_config_check = 0.0 + obj._config_mcp_servers = mcp_servers or {} + + cfg_file = tmp_path / "config.yaml" + cfg_file.write_text("mcp_servers: {}\n") + obj._config_mtime = cfg_file.stat().st_mtime + + obj._reload_mcp = MagicMock() + obj._busy_command = MagicMock() + obj._busy_command.return_value.__enter__ = MagicMock(return_value=None) + obj._busy_command.return_value.__exit__ = MagicMock(return_value=False) + obj._slow_command_status = MagicMock(return_value="reloading...") + + return obj, cfg_file + + +class TestMCPConfigWatch: + + def test_no_change_does_not_reload(self, tmp_path): + """If mtime and mcp_servers unchanged, _reload_mcp is NOT called.""" + obj, cfg_file = _make_cli(tmp_path) + + with patch("hermes_cli.config.get_config_path", return_value=cfg_file): + obj._check_config_mcp_changes() + + obj._reload_mcp.assert_not_called() + + def test_mtime_change_with_same_mcp_servers_does_not_reload(self, tmp_path): + """If file mtime changes but mcp_servers is identical, no reload.""" + import yaml + obj, cfg_file = _make_cli(tmp_path, mcp_servers={"fs": {"command": "npx"}}) + + # Write same mcp_servers but touch the file + cfg_file.write_text(yaml.dump({"mcp_servers": {"fs": {"command": "npx"}}})) + # Force mtime to appear changed + obj._config_mtime = 0.0 + + with patch("hermes_cli.config.get_config_path", return_value=cfg_file): + obj._check_config_mcp_changes() + + obj._reload_mcp.assert_not_called() + + def test_new_mcp_server_triggers_reload(self, tmp_path): + """Adding a new MCP server to config triggers auto-reload.""" + import yaml + obj, cfg_file = _make_cli(tmp_path, mcp_servers={}) + + # Simulate user adding a new MCP server to config.yaml + cfg_file.write_text(yaml.dump({"mcp_servers": {"github": {"url": "https://mcp.github.com"}}})) + obj._config_mtime = 0.0 # force stale mtime + + with patch("hermes_cli.config.get_config_path", return_value=cfg_file): + obj._check_config_mcp_changes() + + obj._reload_mcp.assert_called_once() + + def test_removed_mcp_server_triggers_reload(self, tmp_path): + """Removing an MCP server from config triggers auto-reload.""" + import yaml + obj, cfg_file = _make_cli(tmp_path, mcp_servers={"github": {"url": "https://mcp.github.com"}}) + + # Simulate user removing the server + cfg_file.write_text(yaml.dump({"mcp_servers": {}})) + obj._config_mtime = 0.0 + + with patch("hermes_cli.config.get_config_path", return_value=cfg_file): + obj._check_config_mcp_changes() + + obj._reload_mcp.assert_called_once() + + def test_interval_throttle_skips_check(self, tmp_path): + """If called within CONFIG_WATCH_INTERVAL, stat() is skipped.""" + obj, cfg_file = _make_cli(tmp_path) + obj._last_config_check = time.monotonic() # just checked + + with patch("hermes_cli.config.get_config_path", return_value=cfg_file), \ + patch.object(Path, "stat") as mock_stat: + obj._check_config_mcp_changes() + mock_stat.assert_not_called() + + obj._reload_mcp.assert_not_called() + + def test_missing_config_file_does_not_crash(self, tmp_path): + """If config.yaml doesn't exist, _check_config_mcp_changes is a no-op.""" + obj, cfg_file = _make_cli(tmp_path) + missing = tmp_path / "nonexistent.yaml" + + with patch("hermes_cli.config.get_config_path", return_value=missing): + obj._check_config_mcp_changes() # should not raise + + obj._reload_mcp.assert_not_called()