mcp-server/docs/superpowers/plans/2026-06-30-pluggable-mcp.md

Pluggable MCP Service Registry — Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Refactor the multi-MCP-service wiring in main.py from a hand-maintained hard-coded list into a pluggable, config-gated registry so contributors can add a new MCP service without touching main.py (just append one import-path string), and deployments can opt services in/out via the MCP_SERVICES env var.

Architecture: Move the McpService dataclass from main.py to a new common/mcp_service.py, alongside a BUILTIN_SERVICES list of import paths and a discover_and_filter() helper that imports each, reads the SERVICE attribute, validates uniqueness, and applies the env whitelist. Each service package (spark_executor/, files_mcp/) gains a service.py exporting one SERVICE: McpService instance. main.py's lifespan shrinks to services = discover_and_filter(); for svc in services: mount(svc).

Tech Stack: Python 3.12+, FastAPI 0.138+, Pydantic 2.13+, loguru; stdlib importlib, dataclasses, os. No new dependencies.

Spec: docs/superpowers/specs/2026-06-30-pluggable-mcp-design.md (commit 2ad4a3e on feat/files-mcp).

---

Global Constraints

• Python 3.12+ (per pyproject.toml).
• No new dependencies.
• MCP_SERVICES env var (optional, comma-separated whitelist of service names).
  • Unset: all built-in services mount (backward compat with current behavior).
  • Empty string (MCP_SERVICES=): zero services mount.
  • Subset (e.g. MCP_SERVICES=files): only listed services mount.
  • Unknown name: startup raises with a clear error listing available names.
• common/mcp_service.py is the single source of truth for the registry:
  • McpService dataclass (moved from main.py, byte-for-byte identical fields).
  • BUILTIN_SERVICES: list[str] - every known service's import path.
  • discover_and_filter() -> list[McpService] - does import + validation + env gating in one call.
• Per-service contract: each service.py exports a module-level SERVICE: McpService instance.
• Strict MCP_SERVICES / unset distinction: _filter_by_env checks "MCP_SERVICES" not in os.environ to distinguish unset from empty.
• Duplicate detection: discover_and_filter raises if two services share a name or mount_path.
• No spark_executor/ or files_mcp/ core module changes: only NEW service.py files in each, plus the main.py refactor.
• Module-level file header: Every new Python file starts with # coding=utf-8 and @Time / @Author docstring.
• Branch: feat/files-mcp. All commits land on this branch.
• All work uses uv run for environment consistency.
• Conventions reused from spark_executor/: from common.logging import logger; from typing import TYPE_CHECKING for forward refs.

---

File Structure

Files this plan creates:

common/mcp_service.py                       # McpService + BUILTIN_SERVICES + discover_and_filter
spark_executor/service.py                   # SERVICE = McpService(...)
files_mcp/service.py                        # SERVICE = McpService(...)
tests/unit/test_mcp_service.py              # 12 unit tests
tests/integration/test_main_lifespan.py     # 5 integration tests

Files this plan modifies:

main.py                                     # shrink: remove McpService class + MCP_SERVICES, use discover_and_filter

Files this plan explicitly does NOT touch (must remain byte-for-byte unchanged):

spark_executor/__init__.py
spark_executor/server.py
spark_executor/tools/*
spark_executor/core/*
spark_executor/models.py

files_mcp/__init__.py
files_mcp/server.py
files_mcp/tools/*
files_mcp/core/*
files_mcp/models.py

common/__init__.py
common/factory.py
common/logging.py
common/config.py

tests/conftest.py

---

Task 1: common/mcp_service.py with unit tests

Files:

• Create: common/mcp_service.py
• Create: tests/unit/test_mcp_service.py

Interfaces (this task produces; later tasks consume):

• common.mcp_service.McpService - frozen dataclass with name: str, app: FastAPI, mount_path: str.

• common.mcp_service.BUILTIN_SERVICES: list[str] - module-level constant.

• common.mcp_service._filter_by_env(services: list[McpService]) -> list[McpService] - private helper, public for testing.

• common.mcp_service.discover_and_filter() -> list[McpService] - main entry point.

• Step 1.1: Create common/mcp_service.py

# coding=utf-8
"""
@Time   :2026/6/30
@Author :tao.chen

The MCP service registry: a single place that knows which FastAPI sub-apps
to mount under the root app and how to gate them by the MCP_SERVICES env
var. Each service package declares itself via a `service.py` exporting a
module-level `SERVICE: McpService` instance.
"""
from __future__ import annotations

import importlib
import os
from dataclasses import dataclass
from typing import TYPE_CHECKING

from common.logging import logger

if TYPE_CHECKING:
    from fastapi import FastAPI


@dataclass(frozen=True)
class McpService:
    """One MCP-exposed FastAPI sub-app to be mounted under the root."""

    name: str          # short id used in startup logs and the MCP_SERVICES env whitelist
    app: "FastAPI"     # the sub-app whose routes fastapi-mcp exposes as tools
    mount_path: str    # HTTP path under the root app, e.g. "/spark-executor-mcp"


# Every known MCP service's import path. Add a new service by appending one
# line here and creating <package>/service.py that exports SERVICE.
BUILTIN_SERVICES: list[str] = [
    "spark_executor.service",
    "files_mcp.service",
]


def _filter_by_env(services: list[McpService]) -> list[McpService]:
    """Whitelist by `MCP_SERVICES` env. Unset = no filter (backward compat).
    Empty string = zero services (explicit operator choice, distinct from unset)."""
    if "MCP_SERVICES" not in os.environ:
        return services
    raw = os.environ["MCP_SERVICES"]
    if not raw:
        return []
    allowed = {n.strip() for n in raw.split(",") if n.strip()}
    return [s for s in services if s.name in allowed]


def discover_and_filter() -> list[McpService]:
    """Import each BUILTIN_SERVICES module, read its SERVICE attribute,
    then apply the env whitelist.

    Raises:
        RuntimeError: see `Error Handling` in the spec.
    """
    services: list[McpService] = []
    raw = os.environ.get("MCP_SERVICES")
    requested = {n.strip() for n in raw.split(",") if n.strip()} if raw else set()
    origins: dict[str, str] = {}  # service.name -> import_path (for diagnostics)

    for import_path in BUILTIN_SERVICES:
        try:
            module = importlib.import_module(import_path)
        except ImportError as e:
            pkg_name = import_path.rsplit(".", 1)[0]
            if pkg_name in requested or import_path in requested:
                raise RuntimeError(
                    f"MCP service {import_path!r} is requested via MCP_SERVICES "
                    f"but cannot be imported: {e}"
                ) from e
            logger.warning(f"Skipping missing MCP service {import_path!r}: {e}")
            continue
        svc = getattr(module, "SERVICE", None)
        if svc is None:
            raise RuntimeError(
                f"{import_path!r} does not export a SERVICE attribute"
            )
        if not isinstance(svc, McpService):
            raise RuntimeError(
                f"{import_path!r}.SERVICE is not a McpService "
                f"(got {type(svc).__name__})"
            )
        services.append(svc)
        origins[svc.name] = import_path

    # Duplicate detection - surface typos at startup.
    seen_names: dict[str, str] = {}
    seen_mounts: dict[str, str] = {}
    for s in services:
        if s.name in seen_names:
            raise RuntimeError(
                f"duplicate MCP service name {s.name!r}: "
                f"{seen_names[s.name]!r} and {origins[s.name]!r}"
            )
        seen_names[s.name] = origins[s.name]
        if s.mount_path in seen_mounts:
            raise RuntimeError(
                f"duplicate MCP mount_path {s.mount_path!r}: "
                f"{seen_mounts[s.mount_path]!r} and {origins[s.name]!r}"
            )
        seen_mounts[s.mount_path] = origins[s.name]

    filtered = _filter_by_env(services)

    found_names = {s.name for s in services}
    missing = requested - found_names
    if missing:
        raise RuntimeError(
            f"MCP_SERVICES env references unknown service names: "
            f"{sorted(missing)}. Available: {sorted(found_names)}"
        )

    return filtered

• Step 1.2: Create tests/unit/test_mcp_service.py

# coding=utf-8
"""
@Time   :2026/6/30
@Author :tao.chen
"""
import sys
import types

import pytest
from fastapi import FastAPI

from common import mcp_service
from common.mcp_service import McpService, _filter_by_env, discover_and_filter


# --- helpers ----------------------------------------------------------------


def _fake_app(title: str = "Fake") -> FastAPI:
    return FastAPI(title=title)


def _make_module(monkeypatch, name: str, *, service: object = None) -> None:
    """Inject a fake module into sys.modules with the given SERVICE attribute.

    For "broken import" tests, the test author should set BUILTIN_SERVICES
    to a name that is NOT injected into sys.modules, simulating ImportError.
    """
    mod = types.ModuleType(name)
    if service is not None:
        mod.SERVICE = service
    monkeypatch.setitem(sys.modules, name, mod)


@pytest.fixture
def reset_builtin(monkeypatch):
    """Restore BUILTIN_SERVICES to a known good state after each test."""
    monkeypatch.setattr(
        mcp_service, "BUILTIN_SERVICES",
        ["fake_spark_executor", "fake_files"],
    )


@pytest.fixture
def reset_env(monkeypatch):
    """Make sure MCP_SERVICES is unset at the start of each test."""
    monkeypatch.delenv("MCP_SERVICES", raising=False)


# --- TestDiscover ----------------------------------------------------------


class TestDiscover:
    def test_loads_builtin_services_in_order(
        self, monkeypatch, reset_builtin, reset_env
    ):
        a = McpService(name="fake_spark_executor", app=_fake_app(), mount_path="/a-mcp")
        b = McpService(name="fake_files", app=_fake_app(), mount_path="/b-mcp")
        _make_module(monkeypatch, "fake_spark_executor", service=a)
        _make_module(monkeypatch, "fake_files", service=b)
        result = discover_and_filter()
        assert [s.name for s in result] == ["fake_spark_executor", "fake_files"]

    def test_skips_broken_service_if_not_requested(
        self, monkeypatch, reset_builtin, reset_env, caplog
    ):
        # "fake_spark_executor" is not importable; env doesn't mention it.
        # Only "fake_files" is real.
        b = McpService(name="fake_files", app=_fake_app(), mount_path="/b-mcp")
        _make_module(monkeypatch, "fake_files", service=b)
        with caplog.at_level("WARNING", logger="common.logging"):
            result = discover_and_filter()
        assert [s.name for s in result] == ["fake_files"]
        assert any("fake_spark_executor" in r.message for r in caplog.records)

    def test_raises_on_broken_service_if_requested(
        self, monkeypatch, reset_builtin, reset_env
    ):
        # "fake_spark_executor" is not importable AND operator asked for it.
        monkeypatch.setenv("MCP_SERVICES", "fake_spark_executor")
        b = McpService(name="fake_files", app=_fake_app(), mount_path="/b-mcp")
        _make_module(monkeypatch, "fake_files", service=b)
        with pytest.raises(RuntimeError, match="requested via MCP_SERVICES"):
            discover_and_filter()

    def test_raises_when_service_module_lacks_SERVICE(
        self, monkeypatch, reset_builtin, reset_env
    ):
        # Module exists but has no SERVICE attribute.
        _make_module(monkeypatch, "fake_spark_executor", service=None)
        b = McpService(name="fake_files", app=_fake_app(), mount_path="/b-mcp")
        _make_module(monkeypatch, "fake_files", service=b)
        with pytest.raises(RuntimeError, match="does not export a SERVICE"):
            discover_and_filter()

    def test_raises_when_SERVICE_is_wrong_type(
        self, monkeypatch, reset_builtin, reset_env
    ):
        # SERVICE attribute exists but is not a McpService.
        _make_module(monkeypatch, "fake_spark_executor", service="not a McpService")
        b = McpService(name="fake_files", app=_fake_app(), mount_path="/b-mcp")
        _make_module(monkeypatch, "fake_files", service=b)
        with pytest.raises(RuntimeError, match="is not a McpService"):
            discover_and_filter()

    def test_raises_on_duplicate_service_names(
        self, monkeypatch, reset_builtin, reset_env
    ):
        a = McpService(name="dup", app=_fake_app(), mount_path="/a-mcp")
        b = McpService(name="dup", app=_fake_app(), mount_path="/b-mcp")
        _make_module(monkeypatch, "fake_spark_executor", service=a)
        _make_module(monkeypatch, "fake_files", service=b)
        with pytest.raises(RuntimeError, match="duplicate MCP service name"):
            discover_and_filter()

    def test_raises_on_duplicate_mount_paths(
        self, monkeypatch, reset_builtin, reset_env
    ):
        a = McpService(name="alpha", app=_fake_app(), mount_path="/same")
        b = McpService(name="beta", app=_fake_app(), mount_path="/same")
        _make_module(monkeypatch, "fake_spark_executor", service=a)
        _make_module(monkeypatch, "fake_files", service=b)
        with pytest.raises(RuntimeError, match="duplicate MCP mount_path"):
            discover_and_filter()


# --- TestFilter (unit-tests the private _filter_by_env directly) ----------


class TestFilter:
    def test_unset_env_returns_all(self, monkeypatch):
        monkeypatch.delenv("MCP_SERVICES", raising=False)
        svcs = [
            McpService(name="a", app=_fake_app(), mount_path="/a"),
            McpService(name="b", app=_fake_app(), mount_path="/b"),
        ]
        assert _filter_by_env(svcs) == svcs

    def test_empty_env_returns_zero(self, monkeypatch):
        monkeypatch.setenv("MCP_SERVICES", "")
        svcs = [
            McpService(name="a", app=_fake_app(), mount_path="/a"),
            McpService(name="b", app=_fake_app(), mount_path="/b"),
        ]
        assert _filter_by_env(svcs) == []

    def test_whitelist_filters_by_name(self, monkeypatch):
        monkeypatch.setenv("MCP_SERVICES", "a")
        svcs = [
            McpService(name="a", app=_fake_app(), mount_path="/a"),
            McpService(name="b", app=_fake_app(), mount_path="/b"),
        ]
        result = _filter_by_env(svcs)
        assert [s.name for s in result] == ["a"]

    def test_whitelist_preserves_bulitin_order(self, monkeypatch):
        # env order: "b,a"; result order must follow svcs order, not env order.
        monkeypatch.setenv("MCP_SERVICES", "b,a")
        svcs = [
            McpService(name="a", app=_fake_app(), mount_path="/a"),
            McpService(name="b", app=_fake_app(), mount_path="/b"),
        ]
        result = _filter_by_env(svcs)
        assert [s.name for s in result] == ["a", "b"]


# --- TestEnvErrors ---------------------------------------------------------


class TestEnvErrors:
    def test_unknown_name_in_env_raises_with_available_list(
        self, monkeypatch, reset_builtin, reset_env
    ):
        a = McpService(name="fake_spark_executor", app=_fake_app(), mount_path="/a-mcp")
        b = McpService(name="fake_files", app=_fake_app(), mount_path="/b-mcp")
        _make_module(monkeypatch, "fake_spark_executor", service=a)
        _make_module(monkeypatch, "fake_files", service=b)
        monkeypatch.setenv("MCP_SERVICES", "bogus")
        with pytest.raises(RuntimeError) as exc_info:
            discover_and_filter()
        msg = str(exc_info.value)
        assert "bogus" in msg
        assert "fake_files" in msg
        assert "fake_spark_executor" in msg

• Step 1.3: Run the test to verify it passes

Run: uv run pytest tests/unit/test_mcp_service.py -v Expected: 12 PASS

(If a test fails, debug and fix the production code, NOT the test.)

• Step 1.4: Commit

git add common/mcp_service.py tests/unit/test_mcp_service.py
git commit -m "feat(mcp): add pluggable service registry in common/mcp_service

Adds common/mcp_service.py with the McpService dataclass (moved from
main.py), BUILTIN_SERVICES list, _filter_by_env, and discover_and_filter.
MCP_SERVICES env var acts as a whitelist (unset=all, empty=zero).

Adds tests/unit/test_mcp_service.py with 12 cases covering: load order,
broken-import handling (warn vs raise by env), missing/wrong-typed
SERVICE, duplicate name and duplicate mount_path detection, env filter
(unset/empty/subset/order), and unknown-name error.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

---

Task 2: per-service service.py files + main.py refactor

Files:

• Create: spark_executor/service.py
• Create: files_mcp/service.py
• Modify: main.py (replace McpService class + MCP_SERVICES list with discover_and_filter() call; remove the from files_mcp import app and from spark_executor import app imports; update the explanatory comment block)

Interfaces (this task produces; Task 3 consumes):

• spark_executor.service.SERVICE: McpService

• files_mcp.service.SERVICE: McpService

• main.py:app constructed with the same lifespan; lifespan now calls discover_and_filter().

• Step 2.1: Create spark_executor/service.py

# coding=utf-8
"""
@Time   :2026/6/30
@Author :tao.chen

Pluggable manifest for the spark_executor MCP service. The name and
mount_path here are matched against the MCP_SERVICES env whitelist and
the BUILTIN_SERVICES list in common.mcp_service.
"""
from common.mcp_service import McpService
from spark_executor import app

SERVICE: McpService = McpService(
    name="spark_executor",
    app=app,
    mount_path="/spark-executor-mcp",
)

• Step 2.2: Create files_mcp/service.py

# coding=utf-8
"""
@Time   :2026/6/30
@Author :tao.chen

Pluggable manifest for the files MCP service.
"""
from common.mcp_service import McpService
from files_mcp import app

SERVICE: McpService = McpService(
    name="files",
    app=app,
    mount_path="/files-mcp",
)

• Step 2.3: Read the current main.py

Run: cat main.py Confirm: current main.py has the McpService dataclass, the MCP_SERVICES list, the two from ... import app imports at the top, and the lifespan that iterates MCP_SERVICES. (No other code in main.py is affected.)

• Step 2.4: Replace main.py with the refactored version

The full new content of main.py is:

# coding=utf-8
"""
@Time   :2026/6/30
@Author :tao.chen
"""
from contextlib import asynccontextmanager

from fastapi import FastAPI

from common.factory import init_mcp_server
from common.logging import logger
from common.mcp_service import discover_and_filter


# --- MCP service registry ---
#
# Each service package declares itself via a `service.py` that exports a
# `SERVICE: McpService` instance. The list of known service import paths
# lives in `common.mcp_service.BUILTIN_SERVICES`. The lifespan below calls
# `discover_and_filter()` to import each, validate them, apply the
# MCP_SERVICES env whitelist, and mount the survivors.
#
# To add a new MCP service:
#   1. Build a package with a FastAPI app.
#   2. Add `<package>/service.py` exporting `SERVICE: McpService`.
#   3. Append `"<package>.service"` to `BUILTIN_SERVICES` in
#      `common/mcp_service.py`.


def _log_mcp_tools(mcp_server, service) -> None:
    """Log every tool the MCP server exposes, with its description.

    fastapi-mcp appends an auto-generated "### Responses: ..." block to
    every tool description; we strip that here so the startup log stays
    scannable. Each tool is logged on one line: `  N. tool_name  -  desc`.
    """
    tools = mcp_server.tools
    # Single INFO line per service summarizing how many tools it exposes.
    # Per-tool name + description is DEBUG-level so production logs (INFO)
    # stay scannable but operators can flip to DEBUG to see the full list.
    logger.info(f"MCP service {service.name!r}: {len(tools)} tool(s) registered")
    for i, tool in enumerate(tools, 1):
        desc = (tool.description or "").strip()
        if "### Responses" in desc:
            desc = desc.split("### Responses", 1)[0].rstrip()
        desc = " ".join(desc.split())
        logger.debug(f"  {service.name}.{tool.name}  -  {desc}")


@asynccontextmanager
async def lifespan(app: "FastAPI"):
    logger.info(f"Startup {app.title}")
    services = discover_and_filter()
    for svc in services:
        logger.info(
            f"Mounting MCP service {svc.name!r} at {svc.mount_path} "
            f"(source: {svc.app.title} v{svc.app.version})"
        )
        mcp = init_mcp_server(svc.app)
        mcp.mount_http(app, mount_path=svc.mount_path)
        _log_mcp_tools(mcp, svc)
    yield
    logger.info(f"Shutdown {app.title}")


app = FastAPI(title="Main App", lifespan=lifespan)


@app.get("/health")
def health_check():
    return {"status": "ok"}


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

• Step 2.5: Run the FULL test suite to verify the refactor preserves behavior

Run: uv run pytest -v Expected: 327 passed, 50 warnings in 0.9s (or similar - same total as before this plan started; the file-MCP plan ended at 327 tests, and this refactor adds 0 new tests in Task 2)

If a test fails:

• Check that main.py only has the changes from Step 2.4.

• Check that the per-service service.py files (Step 2.1, 2.2) are exactly as written.

• Check that the existing 327 tests still import their targets directly (e.g. from files_mcp.server import app) - they should not be affected by the registry refactor.

• Most likely failure: a missing or wrong import in service.py. Fix the service.py file (NOT the test) and re-run.

• Step 2.6: Commit

git add main.py spark_executor/service.py files_mcp/service.py
git commit -m "refactor(mcp): use discover_and_filter in main.py; add per-service manifests

- main.py: drops the hard-coded McpService class and MCP_SERVICES list.
  Lifespan now calls discover_and_filter() which does import + env
  filtering. Unset MCP_SERVICES reproduces today's behavior exactly.
- spark_executor/service.py: new file exporting SERVICE = McpService(
  name='spark_executor', app=spark_executor.app, mount_path='/spark-executor-mcp').
- files_mcp/service.py: new file exporting SERVICE = McpService(
  name='files', app=files_mcp.app, mount_path='/files-mcp').

The McpService dataclass itself moved to common/mcp_service.py in the
prior commit. 327 existing tests pass unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

---

Task 3: integration tests for the new lifespan behavior

Files:

• Create: tests/integration/test_main_lifespan.py

Strategy: The existing 327 tests verify individual sub-apps in isolation. This task adds 5 integration tests that verify the end-to-end lifespan wiring in main.py - specifically, that the env var correctly gates which services get mounted.

The tests use TestClient(main.app) and mock init_mcp_server in the main namespace to record which mount_paths get mounted. This avoids depending on fastapi-mcp's HTTP response codes (which 404 for bare GETs regardless of mount state - the MCP transport requires an initialize handshake first).

• Step 3.1: Create tests/integration/test_main_lifespan.py

# coding=utf-8
"""
@Time   :2026/6/30
@Author :tao.chen

End-to-end tests for main.app's lifespan: verify that MCP_SERVICES env
var correctly gates which services get mounted. Mocks main.init_mcp_server
to record mount calls rather than relying on fastapi-mcp's HTTP
response codes (which 404 for bare GETs regardless of mount state
because the MCP transport requires an initialize handshake first).
"""
import pytest
from fastapi.testclient import TestClient


def _get_main_app():
    """Lazily import main.app so the conftest autouse fixture has run
    before files_mcp is first imported (via discover_and_filter inside
    the lifespan)."""
    import main
    return main.app


@pytest.fixture
def mock_mount(monkeypatch):
    """Replace main.init_mcp_server with a recorder.

    Returns a list that gets appended with mount_path each time a
    service is mounted. The lifespan inside the TestClient context
    manager triggers all mounts during __enter__.
    """
    mounted: list[str] = []

    def fake_init(sub_app):
        class _FakeMcp:
            def __init__(self, app):
                self.app = app
            def mount_http(self, parent, mount_path):
                mounted.append(mount_path)
        return _FakeMcp(sub_app)

    monkeypatch.setattr("main.init_mcp_server", fake_init)
    return mounted


class TestLifespanMounting:
    def test_default_mounts_both_services(self, monkeypatch, mock_mount):
        monkeypatch.delenv("MCP_SERVICES", raising=False)
        with TestClient(_get_main_app()):
            pass
        # Both mount paths are present, in BUILTIN_SERVICES order.
        assert mock_mount == ["/spark-executor-mcp", "/files-mcp"]

    def test_MCP_SERVICES_files_only_mounts_one(self, monkeypatch, mock_mount):
        monkeypatch.setenv("MCP_SERVICES", "files")
        with TestClient(_get_main_app()):
            pass
        assert mock_mount == ["/files-mcp"]

    def test_MCP_SERVICES_empty_mounts_none(self, monkeypatch, mock_mount):
        monkeypatch.setenv("MCP_SERVICES", "")
        with TestClient(_get_main_app()):
            pass
        assert mock_mount == []

    def test_MCP_SERVICES_unknown_raises_at_startup(self, monkeypatch, mock_mount):
        monkeypatch.setenv("MCP_SERVICES", "bogus")
        with pytest.raises(RuntimeError, match="unknown service names"):
            with TestClient(_get_main_app()):
                pass
        # No mounts attempted
        assert mock_mount == []

    def test_health_endpoint_unaffected_by_mcp_filtering(self, monkeypatch):
        # /health is on the root app, independent of MCP service mounting.
        monkeypatch.delenv("MCP_SERVICES", raising=False)
        with TestClient(_get_main_app()) as c:
            assert c.get("/health").status_code == 200

        monkeypatch.setenv("MCP_SERVICES", "")
        with TestClient(_get_main_app()) as c:
            assert c.get("/health").status_code == 200

• Step 3.2: Run the integration test to verify it passes

Run: uv run pytest tests/integration/test_main_lifespan.py -v Expected: 5 PASS

If a test fails:

• "No module named 'main'" or import-time error: tests/conftest.py is missing the autouse fixture. Verify with cat tests/conftest.py.

• "Service requested via MCP_SERVICES but cannot be imported": FILES_MCP_ROOT is unset when the lifespan runs. Verify the conftest autouse fixture is autouse=True.

• A test expecting mock_mount == ["/spark-executor-mcp", "/files-mcp"] fails: check the order in common.mcp_service.BUILTIN_SERVICES - spark_executor must come first.

• Do NOT change the test to match buggy behavior. Fix the production code (main.py or common/mcp_service.py) to match the spec.

• Step 3.3: Run the FULL test suite to confirm zero regression

Run: uv run pytest -v Expected: 344 passed, 50 warnings in 0.9s (327 + 12 + 5)

(If the warning count is slightly different, that's fine. The test count must be exactly 344.)

• Step 3.4: Smoke test

Create a one-off script scripts/smoke_pluggable_mcp.py (NOT committed):

# coding=utf-8
"""One-off smoke test: verify MCP_SERVICES env gates mounts at runtime."""
import os
import sys

# Set FILES_MCP_ROOT so files_mcp.__init__ can call _init_root().
os.environ.setdefault("FILES_MCP_ROOT", "/tmp/pluggable-mcp-smoke")
os.makedirs(os.environ["FILES_MCP_ROOT"], exist_ok=True)

import importlib
from fastapi.testclient import TestClient
import main


def scenario(env_value, expected_mounts):
    for k in ("MCP_SERVICES",):
        os.environ.pop(k, None)
    if env_value is not None:
        os.environ["MCP_SERVICES"] = env_value
    importlib.reload(main)
    mounted = []
    real_init = main.init_mcp_server
    def recorder(sub_app):
        class _M:
            def __init__(self, app): self.app = app
            def mount_http(self, parent, mount_path):
                mounted.append(mount_path)
        return _M(sub_app)
    main.init_mcp_server = recorder
    try:
        with TestClient(main.app):
            pass
    finally:
        main.init_mcp_server = real_init
    label = f"MCP_SERVICES={env_value!r}"
    ok = mounted == expected_mounts
    print(f"  {label}: mounted={mounted}  expected={expected_mounts}  {'OK' if ok else 'FAIL'}")
    return ok


print("Smoke test: pluggable MCP service registry")
ok = True
ok &= scenario(None, ["/spark-executor-mcp", "/files-mcp"])  # unset -> all
ok &= scenario("files", ["/files-mcp"])                        # subset
ok &= scenario("", [])                                         # empty -> zero
try:
    ok &= scenario("bogus", [])                                # unknown -> raise
    print("  bogus: expected RuntimeError but got none  FAIL")
    ok = False
except RuntimeError as e:
    print(f"  bogus: raised RuntimeError as expected  OK  ({e})")

sys.exit(0 if ok else 1)

Run: uv run python scripts/smoke_pluggable_mcp.py Expected: 4 OK lines + 1 OK for the bogus case, exit code 0.

Then delete the smoke script: rm scripts/smoke_pluggable_mcp.py (and rmdir scripts if empty).

• Step 3.5: Commit

git add tests/integration/test_main_lifespan.py
git commit -m "test(mcp): integration tests for pluggable lifespan wiring

Adds tests/integration/test_main_lifespan.py with 5 cases that exercise
the end-to-end main.app lifespan:

- default (no MCP_SERVICES) mounts both services in BUILTIN order
- MCP_SERVICES=files mounts only files
- MCP_SERVICES='' mounts zero
- MCP_SERVICES=bogus raises RuntimeError at startup
- /health on the root app is unaffected by MCP filtering

Tests use TestClient and mock main.init_mcp_server to record mount
calls, since fastapi-mcp's HTTP endpoints 404 for bare GETs regardless
of mount state (the MCP transport requires an initialize handshake
before any tool call). Mocking avoids that coupling.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"

---

Self-Review Checklist

Run mentally before declaring the plan done:

1. Spec coverage - every spec requirement has a task:

   • McpService dataclass -> Task 1
   • BUILTIN_SERVICES constant -> Task 1
   • discover_and_filter() with all error modes -> Task 1
   • _filter_by_env distinguishing unset vs empty -> Task 1
   • Duplicate detection (name + mount_path) -> Task 1
   • Per-service service.py manifests -> Task 2 (spark_executor) + Task 2 (files_mcp)
   • main.py refactor (no more hard-coded list) -> Task 2
   • Smoke test (no env, with env, unknown name) -> Task 3
   • 0 changes to spark_executor core / files_mcp core / pyproject / uv.lock -> enforced in Step 2.5 verification + Global Constraints
   • 327 + 17 = 344 tests pass -> Task 3 Step 3.3 acceptance

2. Placeholder scan - no TBD / TODO / "implement later" / "see above" in the plan.

3. Type / name consistency -

   • McpService(name: str, app: FastAPI, mount_path: str) defined in Task 1, used in Task 1 (helpers), Task 2 (manifests), Task 3 (no direct use).
   • BUILTIN_SERVICES: list[str] defined in Task 1, modified in Task 1 tests via monkeypatch.setattr (resets to known good list), never modified in Task 2 or Task 3.
   • _filter_by_env defined in Task 1, tested directly in Task 1, used by discover_and_filter in Task 1.
   • discover_and_filter defined in Task 1, used by main.py lifespan in Task 2, mocked indirectly in Task 3 via main.init_mcp_server (which discover_and_filter -> lifespan -> init_mcp_server calls).
   • SERVICE attribute name consistent across both manifests (Task 2) and discover_and_filter (Task 1 uses getattr(module, "SERVICE", None)).
   • mount_path strings match exactly between service.py files and historical MCP_SERVICES list (/spark-executor-mcp, /files-mcp).

---

Acceptance Criteria

• All 3 task checklists complete.
• uv run pytest -v shows exactly 344 tests passing.
• git diff main --name-only shows the expected scope:
  • common/mcp_service.py (new)
  • spark_executor/service.py (new)
  • files_mcp/service.py (new)
  • main.py (modified)
  • tests/unit/test_mcp_service.py (new)
  • tests/integration/test_main_lifespan.py (new)
• pyproject.toml / uv.lock unchanged.
• All spark_executor/ and files_mcp/ core files (everything except the new service.py per package) are byte-for-byte unchanged.
• Smoke test in Task 3 Step 3.4 prints 4 OK + 1 OK for the bogus case, exits 0.
• Smoke script is deleted (no scripts/ in git status).