fix(mcp): isolate large schemas behind catalog profiles

DEVLOG.md

@@ -22,6 +22,31 @@ Removed the deployment-specific `victor`→`default` cron owner alias and restor
 - Profile tests plus multiplex secret-scope regression → 56 passed.
 - Full `tests/cron` with an isolated temporary `HERMES_HOME` after rebasing onto the latest `origin/axiom` → 702 passed, 2 warnings.
 - Live checks: the shared registry remained stable at 8 Victor / 9 Sentinel rows across a full scheduler minute after all six profile gateways reloaded; Victor sees exactly its eight jobs and explicit default sees zero. The weekly-backup job ran through Victor's profile-local launcher and created a 3.5 GB archive; the Relay watcher ran silently without a script-resolution failure.
+## 2026-07-14 — Isolate large MCP schemas behind catalog profiles
+
+### Summary
+
+Separated MCP connectivity from model-visible schema exposure so large
+catalog-capable integrations no longer inject their full workflow vocabulary
+into ordinary sessions by default.
+
+### What changed
+
+- Added `mcp_servers.<name>.exposure` (`auto`, `catalog`, `direct`, `off`) and
+  optional platform scoping through `expose_on`.
+- Large servers with a complete three-operation catalog bridge now expose only
+  that bridge by default; deferred operations live in an explicit
+  `<server>-direct` session/tool profile.
+- Hardened platform MCP allowlists and `no_mcp` so direct profiles cannot
+  bypass platform scope or accidentally pull in unrelated servers.
+- Added provider-facing schema-size/vocabulary regression coverage and
+  operator configuration documentation.
+
+### Verification
+
+- See AXI-103 PR verification for focused MCP/platform tests and the real
+  provider-facing tool-list before/after measurement.
+
 
 ## 2026-07-13 — Add provider-pluggable music generation plugin
 

hermes_cli/tools_config.py

@@ -1603,22 +1603,36 @@ def _parse_enabled_flag(value, default: bool = True) -> bool:
     return default
 
 
-def enabled_mcp_server_names(config: dict) -> Set[str]:
+def enabled_mcp_server_names(config: dict, platform: Optional[str] = None) -> Set[str]:
     """Names of MCP servers globally enabled in config.yaml.
 
     Shared by the gateway/CLI platform resolver (``_get_platform_tools``) and
     the cron per-job toolset resolver (``cron.scheduler``) so every path agrees
     on MCP membership. A server is enabled unless its config sets an explicitly
     falsey ``enabled`` (per ``_parse_enabled_flag``: false/0/no/off) — a missing
-    flag or an unrecognized value is treated as enabled.
+    flag or an unrecognized value is treated as enabled. When ``platform`` is
+    provided, an optional per-server ``expose_on`` string/list limits schema
+    exposure without disconnecting or de-authenticating the server.
     """
     mcp_servers = (config or {}).get("mcp_servers") or {}
-    return {
-        str(name)
-        for name, server_cfg in mcp_servers.items()
-        if isinstance(server_cfg, dict)
-        and _parse_enabled_flag(server_cfg.get("enabled", True), default=True)
-    }
+    enabled = set()
+    for name, server_cfg in mcp_servers.items():
+        if not isinstance(server_cfg, dict):
+            continue
+        if not _parse_enabled_flag(server_cfg.get("enabled", True), default=True):
+            continue
+        expose_on = server_cfg.get("expose_on")
+        if platform is not None and expose_on is not None:
+            if isinstance(expose_on, str):
+                allowed_platforms = {expose_on}
+            elif isinstance(expose_on, (list, tuple, set)):
+                allowed_platforms = {str(value) for value in expose_on}
+            else:
+                allowed_platforms = set()
+            if platform not in allowed_platforms:
+                continue
+        enabled.add(str(name))
+    return enabled
 
 
 def _exempt_explicit_platform_native(
@@ -1885,21 +1899,38 @@ def _get_platform_tools(
     # If the platform explicitly lists one or more MCP server names, treat that
     # as an allowlist. Otherwise include every globally enabled MCP server.
     # Special sentinel: "no_mcp" in the toolset list disables all MCP servers.
-    enabled_mcp_servers = enabled_mcp_server_names(config)
+    configured_mcp_servers = enabled_mcp_server_names(config)
+    enabled_mcp_servers = enabled_mcp_server_names(config, platform=platform)
+    # Catalog-capable servers may register non-catalog operations in a separate,
+    # opt-in ``<server>-direct`` profile. Treat both aliases as MCP so explicit
+    # allowlists and the no_mcp sentinel remain fail-closed.
+    allowed_direct_profiles = {f"{name}-direct" for name in enabled_mcp_servers}
+    allowed_mcp_profiles = enabled_mcp_servers | allowed_direct_profiles
+    configured_mcp_profiles = configured_mcp_servers | {
+        f"{name}-direct" for name in configured_mcp_servers
+    }
     # Allow "no_mcp" sentinel to opt out of all MCP servers for this platform
     if "no_mcp" in toolset_names:
         explicit_mcp_servers = set()
-        enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers - {"no_mcp"})
+        enabled_toolsets.update(explicit_passthrough - configured_mcp_profiles - {"no_mcp"})
     else:
-        explicit_mcp_servers = explicit_passthrough & enabled_mcp_servers
-        enabled_toolsets.update(explicit_passthrough - enabled_mcp_servers)
+        explicit_mcp_servers = explicit_passthrough & allowed_mcp_profiles
+        enabled_toolsets.update(explicit_passthrough - configured_mcp_profiles)
     if include_default_mcp_servers:
         if explicit_mcp_servers or "no_mcp" in toolset_names:
             enabled_toolsets.update(explicit_mcp_servers)
         else:
             enabled_toolsets.update(enabled_mcp_servers)
     else:
         enabled_toolsets.update(explicit_mcp_servers)
+    # Direct profiles hold deferred operations; the base profile holds the
+    # catalog bridge and must accompany them for a complete surface. Apply this
+    # even when default MCP servers are disabled for a narrow caller.
+    enabled_toolsets.update(
+        profile.removesuffix("-direct")
+        for profile in explicit_mcp_servers
+        if profile.endswith("-direct")
+    )
 
     # Honor agent.disabled_toolsets from config.yaml — allows users to
     # globally suppress specific toolsets (e.g. "memory") across all

tests/hermes_cli/test_tools_config.py

@@ -25,6 +25,7 @@
     gui_toolset_label,
     _visible_providers,
     tools_command,
+    enabled_mcp_server_names,
 )
 
 
@@ -493,6 +494,78 @@ def test_get_platform_tools_includes_enabled_mcp_servers_by_default():
     assert "disabled-server" not in enabled
 
 
+def test_enabled_mcp_servers_honor_platform_exposure_scope():
+    config = {
+        "mcp_servers": {
+            "forge": {
+                "url": "https://forge.example/mcp",
+                "expose_on": ["webhook"],
+            },
+            "shared": {"url": "https://shared.example/mcp"},
+        }
+    }
+
+    assert enabled_mcp_server_names(config, platform="discord") == {"shared"}
+    assert enabled_mcp_server_names(config, platform="webhook") == {"forge", "shared"}
+    # Platform-less callers retain the legacy connectivity view.
+    assert enabled_mcp_server_names(config) == {"forge", "shared"}
+
+
+def test_get_platform_tools_allows_explicit_direct_profile_without_default_catalog_alias():
+    config = {
+        "platform_toolsets": {"webhook": ["web", "forge-direct"]},
+        "mcp_servers": {
+            "forge": {"url": "https://forge.example/mcp", "exposure": "catalog"},
+            "other": {"url": "https://other.example/mcp"},
+        },
+    }
+
+    enabled = _get_platform_tools(config, "webhook")
+
+    assert "forge-direct" in enabled
+    # Direct profiles include the catalog bridge plus the deferred operations.
+    assert "forge" in enabled
+    assert "other" not in enabled
+
+    narrow_enabled = _get_platform_tools(
+        config, "webhook", include_default_mcp_servers=False
+    )
+    assert {"forge", "forge-direct"}.issubset(narrow_enabled)
+    assert "other" not in narrow_enabled
+
+
+def test_get_platform_tools_no_mcp_blocks_direct_profiles_too():
+    config = {
+        "platform_toolsets": {"discord": ["web", "forge-direct", "no_mcp"]},
+        "mcp_servers": {
+            "forge": {"url": "https://forge.example/mcp", "exposure": "catalog"},
+        },
+    }
+
+    enabled = _get_platform_tools(config, "discord")
+
+    assert "forge" not in enabled
+    assert "forge-direct" not in enabled
+
+
+def test_explicit_direct_profile_cannot_bypass_expose_on_scope():
+    config = {
+        "platform_toolsets": {"discord": ["web", "forge-direct"]},
+        "mcp_servers": {
+            "forge": {
+                "url": "https://forge.example/mcp",
+                "exposure": "catalog",
+                "expose_on": ["webhook"],
+            },
+        },
+    }
+
+    enabled = _get_platform_tools(config, "discord")
+
+    assert "forge" not in enabled
+    assert "forge-direct" not in enabled
+
+
 def test_get_platform_tools_keeps_enabled_mcp_servers_with_explicit_builtin_selection():
     config = {
         "platform_toolsets": {"cli": ["web", "memory"]},

tests/tools/test_mcp_dynamic_discovery.py

@@ -35,6 +35,127 @@ def test_exposes_live_server_aliases(self, mock_registry):
             assert validate_toolset("my_srv") is True
             assert "mcp__my_srv__my_tool" in resolve_toolset("my_srv")
 
+    def test_large_catalog_server_defaults_to_catalog_only(self, mock_registry):
+        server = MCPServerTask("forge")
+        server._tools = [
+            _make_mcp_tool("catalog.search", "Discover operations"),
+            _make_mcp_tool("catalog.describe", "Describe operations"),
+            _make_mcp_tool("catalog.call", "Invoke an operation"),
+            *[_make_mcp_tool(f"issues.operation_{i}", "Issue workflow") for i in range(25)],
+        ]
+        server.session = MagicMock()
+
+        with patch("tools.registry.registry", mock_registry):
+            registered = _register_server_tools("forge", server, {})
+
+        catalog_names = set(mock_registry.get_tool_names_for_toolset("mcp-forge"))
+        direct_names = set(mock_registry.get_tool_names_for_toolset("mcp-forge-direct"))
+        assert len(registered) == 28  # connected/discovered, even when not exposed by default
+        assert catalog_names == {
+            "mcp__forge__catalog_search",
+            "mcp__forge__catalog_describe",
+            "mcp__forge__catalog_call",
+        }
+        assert len(direct_names) == 25
+        assert mock_registry.get_toolset_alias_target("forge") == "mcp-forge"
+        assert mock_registry.get_toolset_alias_target("forge-direct") == "mcp-forge-direct"
+
+    def test_catalog_exposure_can_be_explicit_for_small_server(self, mock_registry):
+        server = MCPServerTask("small")
+        server._tools = [
+            _make_mcp_tool("catalog.search"),
+            _make_mcp_tool("catalog.describe"),
+            _make_mcp_tool("catalog.call"),
+            _make_mcp_tool("items.list"),
+        ]
+        server.session = MagicMock()
+
+        with patch("tools.registry.registry", mock_registry):
+            _register_server_tools("small", server, {"exposure": "catalog"})
+
+        assert len(mock_registry.get_tool_names_for_toolset("mcp-small")) == 3
+        assert len(mock_registry.get_tool_names_for_toolset("mcp-small-direct")) == 1
+
+    def test_direct_exposure_preserves_legacy_single_toolset(self, mock_registry):
+        server = MCPServerTask("forge")
+        server._tools = [
+            _make_mcp_tool("catalog.search"),
+            _make_mcp_tool("catalog.describe"),
+            _make_mcp_tool("catalog.call"),
+            _make_mcp_tool("runs.open"),
+        ]
+        server.session = MagicMock()
+
+        with patch("tools.registry.registry", mock_registry):
+            _register_server_tools("forge", server, {"exposure": "direct"})
+
+        directly_exposed = set(mock_registry.get_tool_names_for_toolset("mcp-forge"))
+        assert {
+            "mcp__forge__catalog_search",
+            "mcp__forge__catalog_describe",
+            "mcp__forge__catalog_call",
+            "mcp__forge__runs_open",
+        }.issubset(directly_exposed)
+        assert mock_registry.get_tool_names_for_toolset("mcp-forge-direct") == []
+
+    def test_off_exposure_keeps_connection_but_registers_no_schemas(self, mock_registry):
+        server = MCPServerTask("forge")
+        server._tools = [_make_mcp_tool("catalog.search"), _make_mcp_tool("runs.open")]
+        server.session = MagicMock()
+
+        with patch("tools.registry.registry", mock_registry):
+            registered = _register_server_tools("forge", server, {"exposure": "off"})
+
+        assert registered == []
+        assert mock_registry.get_all_tool_names() == []
+
+    def test_catalog_profile_reduces_real_model_tool_definitions(self, mock_registry):
+        """The provider-facing schema assembly sees only catalog tools by default."""
+        import json
+        import model_tools
+
+        server = MCPServerTask("forge")
+        server._tools = [
+            _make_mcp_tool("catalog.search", "Discover operations"),
+            _make_mcp_tool("catalog.describe", "Describe operations"),
+            _make_mcp_tool("catalog.call", "Invoke an operation"),
+            *[
+                _make_mcp_tool(
+                    f"runs.operation_{i}",
+                    "AgentRun mode: EXECUTE | RESEARCH | REVIEW | DISCUSS",
+                )
+                for i in range(25)
+            ],
+        ]
+        server.session = MagicMock()
+
+        with (
+            patch("tools.registry.registry", mock_registry),
+            patch("model_tools.registry", mock_registry),
+            patch("tools.mcp_tool._make_check_fn", return_value=lambda: True),
+        ):
+            _register_server_tools("forge", server, {})
+            model_tools._clear_tool_defs_cache()
+            ordinary = model_tools.get_tool_definitions(
+                enabled_toolsets=["forge"], quiet_mode=True
+            )
+            explicit = model_tools.get_tool_definitions(
+                enabled_toolsets=["forge", "forge-direct"], quiet_mode=True
+            )
+
+        ordinary_text = json.dumps(ordinary, sort_keys=True)
+        explicit_text = json.dumps(explicit, sort_keys=True)
+        ordinary_names = {tool["function"]["name"] for tool in ordinary}
+        assert ordinary_names == {
+            "mcp__forge__catalog_search",
+            "mcp__forge__catalog_describe",
+            "mcp__forge__catalog_call",
+        }
+        assert "AgentRun" not in ordinary_text
+        assert "EXECUTE" not in ordinary_text
+        assert "mcp__forge__runs_operation_0" in explicit_text
+        assert len(ordinary_text.encode()) < len(explicit_text.encode()) / 4
+
 
 class TestRefreshTools:
     """Tests for MCPServerTask._refresh_tools nuke-and-repave cycle."""