MCP tool visibility and reload after mcp_servers config changes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #1474 fix: auto-reload MCP tools when mcp_servers config changes without restart

  merged to main

  files: cli.py, tests/test_cli_mcp_config_watch.py

  Show the diff

  • cli.py

    @@ -3484,6 +3484,56 @@ def _show_insights(self, command: str = "/insights"):
             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 @@ def run(self):
             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 @@ def process_loop():
                         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:

  • 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()