Fix #286: plugin MCP servers join the config lookup/merge

src/plugins/mcp_integration.py

@@ -2,10 +2,13 @@
 
 import logging
 from dataclasses import dataclass, field
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from .types import LoadedPlugin, PluginManifest
 
+if TYPE_CHECKING:
+    from src.services.mcp.types import McpServerConfig
+
 logger = logging.getLogger(__name__)
 
 
@@ -23,6 +26,13 @@ class McpPluginWrapper:
     server_name: str
     tools: list[McpPluginTool] = field(default_factory=list)
     connected: bool = False
+    # #286: the launch config, set at registration time. With it, the
+    # plugin-scope loader (services/mcp/config.get_managed_mcp_configs)
+    # surfaces this server into the config merge — per-name lookup,
+    # /mcp listings, and scope-policy filtering — like every other
+    # scope. None (legacy registrations) keeps the wrapper tools-only
+    # and invisible to the merge.
+    server_config: "McpServerConfig | None" = None
 
 
 _mcp_plugins: dict[str, McpPluginWrapper] = {}
@@ -33,7 +43,9 @@ def wrap_mcp_server_as_plugin(
     tools: list[dict[str, Any]],
     *,
     description: str = "",
+    server_config: "McpServerConfig | None" = None,
 ) -> McpPluginWrapper:
+    server_type = getattr(server_config, "type", None) or "stdio"
     manifest = PluginManifest(
         name=f"mcp-{server_name}",
         description=description or f"MCP server: {server_name}",
@@ -45,7 +57,7 @@ def wrap_mcp_server_as_plugin(
         manifest=manifest,
         source=f"mcp:{server_name}",
         enabled=True,
-        mcp_servers={server_name: {"type": "stdio"}},
+        mcp_servers={server_name: {"type": server_type}},
     )
 
     mcp_tools: list[McpPluginTool] = []
@@ -62,6 +74,7 @@ def wrap_mcp_server_as_plugin(
         server_name=server_name,
         tools=mcp_tools,
         connected=True,
+        server_config=server_config,
     )
 
     _mcp_plugins[server_name] = wrapper

src/services/mcp/config.py

@@ -499,6 +499,16 @@ def get_mcp_configs_by_scope(
 
 
 def get_mcp_config_by_name(name: str) -> ScopedMcpServerConfig | None:
+    # Enterprise lockdown parity with the aggregate path (#286): when a
+    # managed-mcp.json exists, get_all_mcp_configs returns enterprise
+    # servers ONLY — the by-name resolve (reconnect, OAuth flows) must
+    # not be a side door that hands out user/project/local/managed/
+    # dynamic configs the merge excluded (same reasoning as the C7
+    # approval gate below).
+    if _does_enterprise_mcp_config_exist():
+        servers, _ = get_mcp_configs_by_scope("enterprise")
+        return servers.get(name)
+
     # Order: highest-trust → lowest-trust. Enterprise managed wins over
     # user, which wins over project, which wins over local.
     for scope in ("enterprise", "user", "project", "local"):
@@ -573,26 +583,36 @@ def get_dynamic_mcp_configs() -> dict[str, ScopedMcpServerConfig]:
 def get_managed_mcp_configs() -> dict[str, ScopedMcpServerConfig]:
     """Return plugin-provided MCP server configs (``managed`` scope).
 
-    Phase 7 WI-7.4 (gap #9 subset). The integration point with the plugin
-    layer at ``src/plugins/mcp_integration.py`` is the
-    ``McpPluginWrapper`` registry — but as of today, ``McpPluginWrapper``
-    does not carry an ``McpServerConfig`` (only a ``server_name``,
-    ``plugin``, ``tools``, ``connected``). There is therefore nothing to
-    return: the wrapper holds tool metadata, not the launch config.
-
-    Returning ``{}`` here keeps the merge surface stable: callers (the
-    ``get_all_mcp_configs`` aggregator and ``get_mcp_config_by_name``)
-    can ask for managed configs without crashing, and the moment the
-    plugin layer is extended to carry an ``McpServerConfig`` per wrapper,
-    this loader can read it via ``wrapper.config`` (or whatever the
-    extended schema names it) and propagate.
-
-    TODO(Phase 7 follow-up): extend ``McpPluginWrapper`` with a
-    ``server_config: McpServerConfig`` field, set at registration time,
-    and surface it here. Until that lands, plugin-provided MCP servers
-    cannot participate in the per-name lookup or the merge.
+    Phase 7 WI-7.4 (gap #9 subset), wired by #286: reads the
+    ``McpPluginWrapper`` registry at ``src/plugins/mcp_integration.py``
+    and surfaces every wrapper that carries a ``server_config`` into the
+    merge — per-name lookup, the ``get_all_mcp_configs`` aggregator, and
+    ``filter_mcp_servers_by_policy`` (``allow_managed_only_mcp`` counts
+    these as managed) all see them like any other scope. Legacy
+    tools-only registrations (``server_config=None``) stay invisible to
+    the merge, exactly as before.
+
+    ``plugin_source`` records the providing plugin's name so listings
+    and dedup notices can attribute the entry.
     """
-    return {}
+    try:
+        # Lazy import: services/mcp must not import the plugins package
+        # at module level (plugins imports services.mcp.types).
+        from src.plugins.mcp_integration import get_all_mcp_plugins
+    except ImportError:
+        logger.warning("plugin registry unavailable; no managed MCP servers")
+        return {}
+
+    servers: dict[str, ScopedMcpServerConfig] = {}
+    for wrapper in get_all_mcp_plugins():
+        if wrapper.server_config is None:
+            continue
+        servers[wrapper.server_name] = ScopedMcpServerConfig(
+            config=wrapper.server_config,
+            scope="managed",
+            plugin_source=wrapper.plugin.name,
+        )
+    return servers
 
 
 def get_all_mcp_configs() -> tuple[dict[str, ScopedMcpServerConfig], list[ValidationError]]:
@@ -687,6 +707,33 @@ def get_all_mcp_configs() -> tuple[dict[str, ScopedMcpServerConfig], list[Valida
                 f"duplicate of manual server {rec.get('duplicateOf')!r}."
             )
 
+    # #286: plugin/manual dedup — a plugin server whose launch signature
+    # duplicates a manual entry is suppressed so the operator's explicit
+    # config wins (same policy as the claudeai dedup above — including
+    # the disabled-server carve-out: a DISABLED manual entry must not
+    # suppress its plugin twin, or disabling the manual copy would leave
+    # the user with zero working servers).
+    if managed_servers:
+        managed_only = {k: v for k, v in merged.items() if v.scope == "managed"}
+        manual_only = {
+            k: v
+            for k, v in merged.items()
+            if v.scope in ("user", "project", "local")
+            and not is_mcp_server_disabled(k)
+        }
+        kept_managed, suppressed_plugin = dedup_plugin_mcp_servers(
+            managed_only, manual_only
+        )
+        merged = {
+            **{k: v for k, v in merged.items() if v.scope != "managed"},
+            **kept_managed,
+        }
+        for rec in suppressed_plugin:
+            dedup_notice_strings.append(
+                f"Plugin MCP server {rec.get('name')!r} suppressed; "
+                f"duplicate of {rec.get('duplicateOf')!r}."
+            )
+
     filtered, notices = filter_mcp_servers_by_policy(merged)
 
     all_errors = (

tests/test_plugin_mcp_config.py

@@ -0,0 +1,210 @@
+"""#286 — plugin-provided MCP servers participate in the config merge.
+
+`McpPluginWrapper.server_config` (set at registration) flows through
+`get_managed_mcp_configs` into the aggregator, the per-name lookup, and
+the scope-policy filter like every other scope. Legacy tools-only
+registrations stay invisible to the merge.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from src.plugins.mcp_integration import (
+    clear_mcp_plugins,
+    wrap_mcp_server_as_plugin,
+)
+from src.services.mcp.config import (
+    filter_mcp_servers_by_policy,
+    get_all_mcp_configs,
+    get_managed_mcp_configs,
+    get_mcp_config_by_name,
+)
+from src.services.mcp.config import get_mcp_configs_by_scope as _real_by_scope
+from src.services.mcp.types import McpStdioServerConfig
+
+
+@pytest.fixture(autouse=True)
+def _fresh_plugin_registry(tmp_path, monkeypatch):
+    """Hermetic environment: empty plugin registry, isolated config
+    dirs (these are the suite's first end-to-end get_all_mcp_configs
+    tests — a developer's real ~/.claude or /etc/claude must not flip
+    assertions), and a cleared enterprise-exists cache (a process-wide
+    latch nothing else resets)."""
+    from src.services.mcp.config import clear_enterprise_config_cache
+
+    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path / "cc"))
+    monkeypatch.setenv("CLAUDE_MANAGED_CONFIG_DIR", str(tmp_path / "managed"))
+    clear_enterprise_config_cache()
+    clear_mcp_plugins()
+    yield
+    clear_enterprise_config_cache()
+    clear_mcp_plugins()
+
+
+_CONFIG = McpStdioServerConfig(command="plugin-server", args=["--serve"])
+
+
+class TestManagedLoader:
+    def test_registration_with_config_surfaces_in_managed_scope(self):
+        wrap_mcp_server_as_plugin(
+            "my-plugin-server", [], server_config=_CONFIG
+        )
+        managed = get_managed_mcp_configs()
+        assert "my-plugin-server" in managed
+        scoped = managed["my-plugin-server"]
+        assert scoped.scope == "managed"
+        assert scoped.config is _CONFIG
+        assert scoped.plugin_source == "mcp-my-plugin-server"
+
+    def test_legacy_tools_only_registration_stays_invisible(self):
+        wrap_mcp_server_as_plugin("legacy", [{"name": "t"}])
+        assert get_managed_mcp_configs() == {}
+
+    def test_server_type_reflected_in_loaded_plugin(self):
+        from src.services.mcp.types import McpSSEServerConfig
+
+        wrapper = wrap_mcp_server_as_plugin(
+            "sse-server",
+            [],
+            server_config=McpSSEServerConfig(url="https://example.com/sse"),
+        )
+        assert wrapper.plugin.mcp_servers == {"sse-server": {"type": "sse"}}
+
+
+class TestMergeParticipation:
+    def test_appears_in_aggregate_and_by_name(self):
+        wrap_mcp_server_as_plugin("merged-in", [], server_config=_CONFIG)
+        servers, _errors = get_all_mcp_configs()
+        assert "merged-in" in servers
+        assert servers["merged-in"].scope == "managed"
+
+        scoped = get_mcp_config_by_name("merged-in")
+        assert scoped is not None
+        assert scoped.scope == "managed"
+        assert scoped.config is _CONFIG
+
+    def test_manual_same_name_overrides_plugin(self):
+        wrap_mcp_server_as_plugin("shadowed", [], server_config=_CONFIG)
+        from src.services.mcp.types import ScopedMcpServerConfig
+
+        manual = ScopedMcpServerConfig(
+            config=McpStdioServerConfig(command="manual-bin"),
+            scope="user",
+        )
+        def _by_scope(scope):
+            if scope == "user":
+                return {"shadowed": manual}, []
+            return _real_by_scope(scope)
+
+        with patch(
+            "src.services.mcp.config.get_mcp_configs_by_scope",
+            side_effect=_by_scope,
+        ):
+            servers, _errors = get_all_mcp_configs()
+        assert servers["shadowed"].scope == "user"
+        assert servers["shadowed"].config.command == "manual-bin"
+
+    def test_signature_duplicate_of_manual_is_suppressed_with_notice(self):
+        # Same launch signature under a DIFFERENT name: the operator's
+        # explicit entry wins; the plugin copy is suppressed + noticed.
+        wrap_mcp_server_as_plugin(
+            "plugin-name",
+            [],
+            server_config=McpStdioServerConfig(command="same-bin", args=["-x"]),
+        )
+        from src.services.mcp.types import ScopedMcpServerConfig
+
+        manual = ScopedMcpServerConfig(
+            config=McpStdioServerConfig(command="same-bin", args=["-x"]),
+            scope="user",
+        )
+        def _by_scope(scope):
+            if scope == "user":
+                return {"manual-name": manual}, []
+            return _real_by_scope(scope)
+
+        with patch(
+            "src.services.mcp.config.get_mcp_configs_by_scope",
+            side_effect=_by_scope,
+        ):
+            servers, errors = get_all_mcp_configs()
+        assert "manual-name" in servers
+        assert "plugin-name" not in servers
+        assert any(
+            "plugin-name" in e.message and "manual-name" in e.message
+            for e in errors
+        )
+
+
+class TestPolicyParticipation:
+    def test_allow_managed_only_keeps_plugin_servers(self):
+        wrap_mcp_server_as_plugin("kept", [], server_config=_CONFIG)
+        managed = get_managed_mcp_configs()
+        from src.services.mcp.types import ScopedMcpServerConfig
+
+        user_entry = ScopedMcpServerConfig(
+            config=McpStdioServerConfig(command="user-bin"), scope="user"
+        )
+        with patch(
+            "src.services.mcp.config._safe_load_settings",
+            return_value=SimpleNamespace(
+                extra={"allow_managed_only_mcp": True}
+            ),
+        ):
+            filtered, notices = filter_mcp_servers_by_policy(
+                {**managed, "user-server": user_entry}
+            )
+        assert "kept" in filtered
+        assert "user-server" not in filtered
+
+    def test_disabled_manual_twin_does_not_suppress_plugin(self):
+        # The claudeai-dedup carve-out applies here too: a DISABLED
+        # manual entry must not suppress its plugin twin, or disabling
+        # the manual copy would leave zero working servers.
+        from src.services.mcp.config import set_mcp_server_enabled
+        from src.services.mcp.types import ScopedMcpServerConfig
+
+        wrap_mcp_server_as_plugin(
+            "plugin-twin",
+            [],
+            server_config=McpStdioServerConfig(command="twin-bin", args=["-y"]),
+        )
+        manual = ScopedMcpServerConfig(
+            config=McpStdioServerConfig(command="twin-bin", args=["-y"]),
+            scope="user",
+        )
+
+        def _by_scope(scope):
+            if scope == "user":
+                return {"manual-twin": manual}, []
+            return _real_by_scope(scope)
+
+        set_mcp_server_enabled("manual-twin", False)
+        try:
+            with patch(
+                "src.services.mcp.config.get_mcp_configs_by_scope",
+                side_effect=_by_scope,
+            ):
+                servers, _errors = get_all_mcp_configs()
+        finally:
+            set_mcp_server_enabled("manual-twin", True)
+        assert "plugin-twin" in servers
+
+
+class TestEnterpriseLockdownByName:
+    def test_by_name_honors_enterprise_short_circuit(self):
+        # When a managed-mcp.json exists, the aggregate returns
+        # enterprise-only; the by-name resolve must not be a side door
+        # that still hands out plugin configs (#286).
+        wrap_mcp_server_as_plugin("locked-out", [], server_config=_CONFIG)
+        with patch(
+            "src.services.mcp.config._does_enterprise_mcp_config_exist",
+            return_value=True,
+        ), patch(
+            "src.services.mcp.config.get_mcp_configs_by_scope",
+            return_value=({}, []),
+        ):
+            assert get_mcp_config_by_name("locked-out") is None