# 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`**

```python
# 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`**

```python
# 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**

```bash
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`**

```python
# 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`**

```python
# 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:

```python
# 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**

```bash
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_path`s 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`**

```python
# 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):

```python
# 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**

```bash
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`).