feat(hermes): Add Hermes Mnemosyne tool whitelist

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [SemVer](https://semver.org/) starting from v3.1.2.
 
 ## [Unreleased]
 
+### Added
+
+- Hermes Mnemosyne providers can now restrict exposed tools with the optional
+  `memory.mnemosyne.tools` allowlist while preserving memory context/prefetch
+  behavior.
+
 ## [3.10.1] — 2026-06-22
 
 ### Security

hermes_memory_provider/__init__.py

@@ -1391,6 +1391,39 @@ def _read_config_key(self, key: str) -> Any:
         except Exception:
             return None
 
+
+    def _configured_tool_schemas(self) -> List[Dict[str, Any]]:
+        """Return schemas filtered by memory.mnemosyne.tools, if configured.
+
+        ``tools`` omitted/None preserves the historical behavior and exposes all
+        Mnemosyne tools. ``tools: []`` exposes no tools while still allowing the
+        provider's memory context/prefetch surface to initialize. Unknown names
+        fail loudly so operators catch typos during Hermes startup instead of
+        silently losing tools.
+        """
+        configured = self._read_config_key("tools")
+        if configured is None:
+            return list(ALL_TOOL_SCHEMAS)
+        if isinstance(configured, str):
+            configured = [name.strip() for name in configured.replace(",", "\n").split("\n") if name.strip()]
+        if not isinstance(configured, list):
+            raise ValueError("memory.mnemosyne.tools must be a list of tool names")
+
+        available = {schema["name"]: schema for schema in ALL_TOOL_SCHEMAS}
+        unknown = [name for name in configured if name not in available]
+        if unknown:
+            known = ", ".join(sorted(available))
+            bad = ", ".join(str(name) for name in unknown)
+            raise ValueError(f"Unknown Mnemosyne tool(s) in memory.mnemosyne.tools: {bad}. Known tools: {known}")
+        return [available[name] for name in configured]
+
+    def _configured_tool_names(self) -> Set[str]:
+        return {schema["name"] for schema in self._configured_tool_schemas()}
+
+    def has_tool(self, tool_name: str) -> bool:
+        """Return whether a tool is currently exposed by this provider."""
+        return tool_name in self._configured_tool_names()
+
     def _reflection_skip_response(self, reason: str, trigger: str) -> Dict[str, Any]:
         """Structured skip payload for reflection/sleep guardrails."""
         return {
@@ -1430,6 +1463,7 @@ def get_config_schema(self) -> List[Dict[str, Any]]:
             {"key": "skip_contexts", "description": "Agent contexts where Mnemosyne should skip initialization. Comma-separated list. Defaults to 'cron,flush,subagent,background,skill_loop'. Set to empty string to enable all contexts. Also configurable via MNEMOSYNE_SKIP_CONTEXTS env var.", "default": "cron,flush,subagent,background,skill_loop"},
             {"key": "sync_roles", "description": "Conversation roles to autosave in sync_turn(). List of role names: 'user', 'assistant'. Default ['user', 'assistant'] saves both. Set to ['user'] for user turns only, or [] to disable conversation autosave entirely. Does not affect explicit mnemosyne_remember calls. Identity signal capture is gated by user sync — excluding 'user' also disables identity extraction. Also configurable via MNEMOSYNE_SYNC_ROLES env var.", "default": ["user", "assistant"]},
             {"key": "default_scope", "description": "Default scope for remember() calls when not explicitly specified. 'session' (default) limits memories to the current session. 'global' persists memories across sessions.", "choices": ["session", "global"], "default": "session"},
+            {"key": "tools", "description": "Optional list of Mnemosyne tool names to expose to Hermes. Omit or set null to expose all tools. Set [] to expose no tools while keeping memory context/prefetch enabled. Unknown names raise a clear startup/config error.", "default": None},
         ]
 
     def save_config(self, values: Dict[str, Any], hermes_home: str) -> None:
@@ -1961,10 +1995,15 @@ def _sleep_isolated():
             pass
 
     def get_tool_schemas(self) -> List[Dict[str, Any]]:
-        """Return tool schemas — static, do not depend on initialization state."""
-        return list(ALL_TOOL_SCHEMAS)
+        """Return configured tool schemas; independent of Beam initialization state."""
+        return self._configured_tool_schemas()
 
     def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str:
+        try:
+            if not self.has_tool(tool_name):
+                return json.dumps({"error": f"Unknown Mnemosyne tool: {tool_name}"})
+        except ValueError as exc:
+            return json.dumps({"error": str(exc)})
         if tool_name == "mnemosyne_sleep" and self._reflect_disabled_for_cron and (self._agent_context or "").strip().lower() == "cron":
             return json.dumps(self._reflection_skip_response("reflect_disabled_for_cron", "tool"))
         if not self._beam:

integrations/hermes/src/mnemosyne_hermes/__init__.py

@@ -631,6 +631,39 @@ def _read_config_key(self, key: str) -> Any:
         except Exception:
             return None
 
+
+    def _configured_tool_schemas(self) -> List[Dict[str, Any]]:
+        """Return schemas filtered by memory.mnemosyne.tools, if configured.
+
+        ``tools`` omitted/None preserves the historical behavior and exposes all
+        Mnemosyne tools. ``tools: []`` exposes no tools while still allowing the
+        provider's memory context/prefetch surface to initialize. Unknown names
+        fail loudly so operators catch typos during Hermes startup instead of
+        silently losing tools.
+        """
+        configured = self._read_config_key("tools")
+        if configured is None:
+            return list(ALL_TOOL_SCHEMAS)
+        if isinstance(configured, str):
+            configured = [name.strip() for name in configured.replace(",", "\n").split("\n") if name.strip()]
+        if not isinstance(configured, list):
+            raise ValueError("memory.mnemosyne.tools must be a list of tool names")
+
+        available = {schema["name"]: schema for schema in ALL_TOOL_SCHEMAS}
+        unknown = [name for name in configured if name not in available]
+        if unknown:
+            known = ", ".join(sorted(available))
+            bad = ", ".join(str(name) for name in unknown)
+            raise ValueError(f"Unknown Mnemosyne tool(s) in memory.mnemosyne.tools: {bad}. Known tools: {known}")
+        return [available[name] for name in configured]
+
+    def _configured_tool_names(self) -> Set[str]:
+        return {schema["name"] for schema in self._configured_tool_schemas()}
+
+    def has_tool(self, tool_name: str) -> bool:
+        """Return whether a tool is currently exposed by this provider."""
+        return tool_name in self._configured_tool_names()
+
     def _reflection_skip_response(self, reason: str, trigger: str) -> Dict[str, Any]:
         """Structured skip payload for reflection/sleep guardrails."""
         return {
@@ -670,6 +703,7 @@ def get_config_schema(self) -> List[Dict[str, Any]]:
             {"key": "skip_contexts", "description": "Agent contexts where Mnemosyne should skip initialization. Comma-separated list. Defaults to 'cron,flush,subagent,background,skill_loop'. Set to empty string to enable all contexts. Also configurable via MNEMOSYNE_SKIP_CONTEXTS env var.", "default": "cron,flush,subagent,background,skill_loop"},
             {"key": "sync_roles", "description": "Conversation roles to autosave in sync_turn(). List of role names: 'user', 'assistant'. Default ['user', 'assistant'] saves both. Set to ['user'] for user turns only, or [] to disable conversation autosave entirely. Does not affect explicit mnemosyne_remember calls. Identity signal capture is gated by user sync — excluding 'user' also disables identity extraction. Also configurable via MNEMOSYNE_SYNC_ROLES env var.", "default": ["user", "assistant"]},
             {"key": "default_scope", "description": "Default scope for remember() calls when not explicitly specified. 'session' (default) limits memories to the current session. 'global' persists memories across sessions.", "choices": ["session", "global"], "default": "session"},
+            {"key": "tools", "description": "Optional list of Mnemosyne tool names to expose to Hermes. Omit or set null to expose all tools. Set [] to expose no tools while keeping memory context/prefetch enabled. Unknown names raise a clear startup/config error.", "default": None},
         ]
 
     def save_config(self, values: Dict[str, Any], hermes_home: str) -> None:
@@ -1081,10 +1115,15 @@ def _maybe_auto_sleep(self) -> None:
             pass
 
     def get_tool_schemas(self) -> List[Dict[str, Any]]:
-        """Return tool schemas — static, do not depend on initialization state."""
-        return list(ALL_TOOL_SCHEMAS)
+        """Return configured tool schemas; independent of Beam initialization state."""
+        return self._configured_tool_schemas()
 
     def handle_tool_call(self, tool_name: str, args: Dict[str, Any], **kwargs) -> str:
+        try:
+            if not self.has_tool(tool_name):
+                return json.dumps({"error": f"Unknown Mnemosyne tool: {tool_name}"})
+        except ValueError as exc:
+            return json.dumps({"error": str(exc)})
         if tool_name == "mnemosyne_sleep" and self._reflect_disabled_for_cron and (self._agent_context or "").strip().lower() == "cron":
             return json.dumps(self._reflection_skip_response("reflect_disabled_for_cron", "tool"))
         if not self._beam:

tests/test_hermes_provider_parity.py

@@ -22,14 +22,28 @@ def _drop_modules(prefix: str) -> None:
 
 def _import_module(package: str, import_root: Path):
     _drop_modules(package)
-    sys.path.insert(0, str(import_root))
+    saved_mnemosyne_modules = {
+        name: module for name, module in sys.modules.items()
+        if name == "mnemosyne" or name.startswith("mnemosyne.")
+    }
+    _drop_modules("mnemosyne")
+    inserted = [str(import_root)]
+    if import_root != PROJECT_ROOT:
+        inserted.append(str(PROJECT_ROOT))
+    for path in reversed(inserted):
+        sys.path.insert(0, path)
     try:
         return importlib.import_module(package)
     finally:
-        try:
-            sys.path.remove(str(import_root))
-        except ValueError:
-            pass
+        for path in inserted:
+            try:
+                sys.path.remove(path)
+            except ValueError:
+                pass
+        for name in list(sys.modules):
+            if name == "mnemosyne" or name.startswith("mnemosyne."):
+                sys.modules.pop(name, None)
+        sys.modules.update(saved_mnemosyne_modules)
 
 
 @pytest.fixture(scope="module")
@@ -49,6 +63,31 @@ def _config_schema(module):
     return {entry["key"]: entry for entry in provider.get_config_schema()}
 
 
+def _write_mnemosyne_config(hermes_home: Path, tools) -> None:
+    if tools is None:
+        body = "memory:\n  provider: mnemosyne\n  mnemosyne: {}\n"
+    else:
+        rendered_tools = "\n".join(f"      - {tool}" for tool in tools)
+        body = (
+            "memory:\n"
+            "  provider: mnemosyne\n"
+            "  mnemosyne:\n"
+            "    tools:\n"
+            f"{rendered_tools}\n"
+        )
+    (hermes_home / "config.yaml").write_text(body)
+
+
+def _schema_names(provider) -> list[str]:
+    return [schema["name"] for schema in provider.get_tool_schemas()]
+
+
+def _provider_for_config(module, hermes_home: Path):
+    provider = module.MnemosyneMemoryProvider()
+    provider._hermes_home = str(hermes_home)
+    return provider
+
+
 def _json_stable(value):
     return json.loads(json.dumps(value, sort_keys=True))
 
@@ -77,6 +116,63 @@ def test_provider_config_defaults_match(provider_modules):
     assert root_config["sync_roles"]["default"] == ["user", "assistant"]
     assert root_config["default_scope"]["choices"] == ["session", "global"]
     assert root_config["default_scope"]["default"] == "session"
+    assert root_config["tools"]["default"] is None
+
+
+def test_tool_whitelist_omitted_exposes_all_tools(tmp_path, provider_modules):
+    _write_mnemosyne_config(tmp_path, None)
+
+    observed = {}
+    for name, module in provider_modules.items():
+        provider = _provider_for_config(module, tmp_path)
+        observed[name] = _schema_names(provider)
+
+    all_tools = list(_tool_schemas(provider_modules["hermes_memory_provider"]))
+    assert observed["hermes_memory_provider"] == all_tools
+    assert observed["mnemosyne_hermes"] == all_tools
+
+
+def test_tool_whitelist_filters_schemas_before_routing(tmp_path, provider_modules):
+    allowed = ["mnemosyne_remember", "mnemosyne_recall", "mnemosyne_sleep"]
+    _write_mnemosyne_config(tmp_path, allowed)
+
+    observed = {}
+    for name, module in provider_modules.items():
+        provider = _provider_for_config(module, tmp_path)
+        observed[name] = _schema_names(provider)
+        assert provider.has_tool("mnemosyne_remember") is True
+        assert provider.has_tool("mnemosyne_forget") is False
+        rejected = json.loads(provider.handle_tool_call("mnemosyne_forget", {"memory_id": "x"}))
+        assert rejected == {"error": "Unknown Mnemosyne tool: mnemosyne_forget"}
+
+    assert observed["hermes_memory_provider"] == allowed
+    assert observed["mnemosyne_hermes"] == allowed
+    assert "mnemosyne_forget" not in observed["hermes_memory_provider"]
+    # Hermes builds its tool routing map from exposed schemas; filtered-out
+    # names must therefore be absent from that registration surface.
+    assert "mnemosyne_forget" not in set(observed["mnemosyne_hermes"])
+
+
+def test_tool_whitelist_empty_list_exposes_no_tools(tmp_path, provider_modules):
+    (tmp_path / "config.yaml").write_text(
+        "memory:\n"
+        "  provider: mnemosyne\n"
+        "  mnemosyne:\n"
+        "    tools: []\n"
+    )
+
+    for module in provider_modules.values():
+        provider = _provider_for_config(module, tmp_path)
+        assert provider.get_tool_schemas() == []
+
+
+def test_tool_whitelist_unknown_name_fails_loudly(tmp_path, provider_modules):
+    _write_mnemosyne_config(tmp_path, ["mnemosyne_remember", "mnemosyne_not_real"])
+
+    for module in provider_modules.values():
+        provider = _provider_for_config(module, tmp_path)
+        with pytest.raises(ValueError, match="Unknown Mnemosyne tool.*mnemosyne_not_real"):
+            provider.get_tool_schemas()
 
 
 @pytest.mark.parametrize(
@@ -170,28 +266,52 @@ def test_provider_sync_turn_zero_limit_means_untruncated(monkeypatch, provider_m
     ]
 
 
-def test_provider_persona_tool_dispatch_matches(tmp_path, provider_modules):
-    from mnemosyne.core.beam import BeamMemory
+def _save_mnemosyne_modules():
+    return {
+        name: module for name, module in sys.modules.items()
+        if name == "mnemosyne" or name.startswith("mnemosyne.")
+    }
 
-    observed = {}
-    for name, module in provider_modules.items():
-        db_path = tmp_path / f"{name}.db"
-        beam = BeamMemory(session_id=f"persona-{name}", db_path=str(db_path))
-        beam.conn.execute(
-            "INSERT INTO memoria_persona (tier, topic, content, confidence) "
-            "VALUES (?, ?, ?, ?)",
-            ("long_term", "test", f"persona rule for {name}", 0.9),
-        )
-        beam.conn.commit()
-
-        provider = module.MnemosyneMemoryProvider.__new__(module.MnemosyneMemoryProvider)
-        provider._beam = beam
-        result = json.loads(provider.handle_tool_call("mnemosyne_persona_list", {}))
-        observed[name] = {
-            "status": result.get("status"),
-            "count": result.get("count"),
-            "topics": [row.get("topic") for row in result.get("personas", [])],
-        }
+
+def _restore_mnemosyne_modules(saved_modules):
+    for name in list(sys.modules):
+        if name == "mnemosyne" or name.startswith("mnemosyne."):
+            sys.modules.pop(name, None)
+    sys.modules.update(saved_modules)
+
+
+def test_provider_persona_tool_dispatch_matches(tmp_path, provider_modules):
+    saved_mnemosyne_modules = _save_mnemosyne_modules()
+    _drop_modules("mnemosyne")
+    sys.path.insert(0, str(PROJECT_ROOT))
+    try:
+        from mnemosyne.core.beam import BeamMemory
+
+        observed = {}
+        for name, module in provider_modules.items():
+            db_path = tmp_path / f"{name}.db"
+            beam = BeamMemory(session_id=f"persona-{name}", db_path=str(db_path))
+            beam.conn.execute(
+                "INSERT INTO memoria_persona (tier, topic, content, confidence) "
+                "VALUES (?, ?, ?, ?)",
+                ("long_term", "test", f"persona rule for {name}", 0.9),
+            )
+            beam.conn.commit()
+
+            provider = module.MnemosyneMemoryProvider.__new__(module.MnemosyneMemoryProvider)
+            provider._beam = beam
+            result = json.loads(provider.handle_tool_call("mnemosyne_persona_list", {}))
+            observed[name] = {
+                "status": result.get("status"),
+                "count": result.get("count"),
+                "topics": [row.get("topic") for row in result.get("personas", [])],
+            }
+    finally:
+        try:
+            sys.path.remove(str(PROJECT_ROOT))
+        except ValueError:
+            pass
+        _restore_mnemosyne_modules(saved_mnemosyne_modules)
 
     assert observed["hermes_memory_provider"] == observed["mnemosyne_hermes"]
     assert observed["hermes_memory_provider"] == {