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

Pluggable MCP Service Registry — Design Spec

Date: 2026-06-30 Branch: feat/files-mcp (continues from the file-MCP service work; same branch because the work is logically related and the branch has not been pushed) Status: Design approved; awaiting implementation plan

Goal

Refactor the multi-MCP-service wiring in main.py from a hand-maintained hard-coded list into a pluggable, config-gated registry so that:

1. Contributors can add a new MCP service without editing main.py other than appending a single import-path string. Each service package owns its own manifest; main.py no longer imports the service's FastAPI app.
2. Deployments can opt services in or out per environment via the MCP_SERVICES env var (whitelist). Unset = all built-in services (backward compatible with the current behavior).

Out of scope: third-party pip packages auto-registering via Python entry points. That is a heavier design (separate versioning, namespace packages, install-order semantics) and is not needed for the in-repo services.

Background

The current main.py has:

from files_mcp import app as files_app
from spark_executor import app as spark_executor_app

@dataclass(frozen=True)
class McpService: ...
MCP_SERVICES: list[McpService] = [
    McpService(name="spark_executor", app=spark_executor_app, mount_path="/spark-executor-mcp"),
    McpService(name="files",         app=files_app,         mount_path="/files-mcp"),
]

To add a third service today, the contributor must:

1. Build the service's package.
2. Add a new from <pkg> import app as <pkg>_app import to main.py.
3. Add a new McpService(...) entry to MCP_SERVICES.

Steps 2 and 3 are mechanical and add no value — they only couple main.py to every service package's identity. This spec removes that coupling.

Non-Goals

• Third-party pip-package registration via Python entry points. Not needed for in-repo services; would add namespace/versioning concerns.
• Hot-reload of services at runtime. Services are discovered at startup in the lifespan once. Re-discovery mid-process is out of scope.
• Per-service instance count / sharding. Each service has exactly one instance. Multi-tenant sharding is Stage 3.
• A web UI for toggling services. Configuration is env-var only.

Architecture

common/mcp_service.py            # NEW: McpService + BUILTIN_SERVICES + discover_and_filter
  - McpService dataclass (moved from main.py)
  - BUILTIN_SERVICES: list[str] (module-paths, e.g. "files_mcp.service")
  - discover_and_filter() -> list[McpService]  (does import + env gating in one call)

spark_executor/service.py        # NEW: exports SERVICE = McpService(...)
files_mcp/service.py             # NEW: exports SERVICE = McpService(...)

main.py                          # MODIFIED: shrinks; lifespan calls discover_and_filter()

Why a separate common/mcp_service.py (not a new top-level package)?

• The common/ package already exists. Adding one file there is the lowest-disruption option.
• A new top-level package (mcp_common/) would be a larger refactor with no functional benefit at this scale.
• The dependency graph stays acyclic: common/mcp_service.py imports nothing project-specific; both service.py files and main.py import from it.

Why a per-package service.py (not a MANIFEST.toml)?

• Python import-time module = the manifest. No file-format dance.
• The SERVICE = McpService(...) line is a one-liner the contributor cannot get wrong structurally.
• IDE/type-checker support is automatic; TOML manifests would need a parser and a schema.
• A .toml would force runtime errors for typos; a Python attribute surfaces the error at import time.

Data Model

# common/mcp_service.py
from __future__ import annotations
import importlib
import os
from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from fastapi import FastAPI

from common.logging import logger


@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 env whitelist
    app: "FastAPI"     # the sub-app whose routes fastapi-mcp exposes as tools
    mount_path: str    # HTTP path under the root app


# Module-level constant: every known MCP service's import path.
# Add a new service by appending one line here.
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.

    Failure modes:
      - Import fails AND `MCP_SERVICES` did not request it: WARN, skip.
      - Import fails AND `MCP_SERVICES` requested it: RAISE.
      - Module lacks `SERVICE` attribute: RAISE.
      - `SERVICE` is not a McpService instance: RAISE.
      - `MCP_SERVICES` names a service not in BUILTIN_SERVICES: RAISE.
      - Duplicate `name` across discovered services: RAISE.
      - Duplicate `mount_path` across discovered services: RAISE.
    """
    services: list[McpService] = []
    raw = os.environ.get("MCP_SERVICES")
    requested = {n.strip() for n in raw.split(",") if n.strip()} if raw else set()

    # Track each service's import path for diagnostic messages.
    origins: dict[str, str] = {}  # service.name -> import_path

    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

Per-Service Manifest Contract

Each MCP service package MUST provide a service.py that exports a module-level SERVICE: McpService instance. Example:

# files_mcp/service.py
# coding=utf-8
from files_mcp import app
from common.mcp_service import McpService

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

# spark_executor/service.py
# coding=utf-8
from spark_executor import app
from common.mcp_service import McpService

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

Contract (enforced by discover_and_filter):

• The SERVICE attribute exists and is a McpService instance.
• The app is already constructed (the import in the manifest is what triggers app construction).
• The name and mount_path are unique across all built-in services.

main.py After

# main.py (refactored)
# coding=utf-8
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


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
    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)

Configuration

• MCP_SERVICES (env var, optional): comma-separated whitelist of service names, e.g. MCP_SERVICES=spark_executor,files.
  • Unset: all built-in services are mounted (current behavior).
  • Empty string (MCP_SERVICES=): no services are mounted.
  • Subset (e.g. MCP_SERVICES=files): only listed services are mounted.
  • Unknown name: startup fails with a clear error listing available names.
• No new config files, no pyproject.toml changes, no new dependencies.

Error Handling

Scenario	Behavior	Rationale
MCP_SERVICES unset	All built-in services mount	Backward compat.
MCP_SERVICES="" (empty)	Zero services mount	Operator explicit choice.
MCP_SERVICES=files (subset)	Only files mounts	Per-deployment opt-in.
MCP_SERVICES=files,bogus	Startup raises; error lists available names	Typo guard.
BUILTIN path import fails, not requested in env	WARN, skip	Tolerable.
BUILTIN path import fails, requested in env	Startup raises	Operator asked for it, we owe it.
service.py lacks SERVICE attribute	Startup raises	Service package bug.
SERVICE is wrong type	Startup raises	Service package bug.
Duplicate name across services	Startup raises	Service package collision.
Duplicate mount_path across services	Startup raises	Service package collision.

Data Flow

process start
  -> main module import
  -> main.app created with lifespan
  -> uvicorn invokes lifespan:
      1. discover_and_filter()
         a. for each path in BUILTIN_SERVICES:
            - importlib.import_module(path)        <- may raise ImportError
            - read module.SERVICE                   <- may be missing / wrong type
         b. check for duplicate name / mount_path   <- raises if any
         c. apply MCP_SERVICES env whitelist
         d. check that env names match a discovered service
         e. return list[McpService]
      2. for svc in result:
         a. log "Mounting ..."
         b. mcp = init_mcp_server(svc.app)
         c. mcp.mount_http(app, mount_path=svc.mount_path)
         d. _log_mcp_tools(mcp, svc)
  -> app serves requests

Backward Compatibility

• No existing tool / route / model / test changes. The refactor only moves McpService out of main.py and adds a manifest module per service.
• Unset MCP_SERVICES = identical behavior to today. A deployment that doesn't set the env var will mount both spark_executor and files exactly as before.
• Existing 327 tests pass unchanged (no test code touches main.py's service list; tests use direct TestClient on the sub-apps).
• gunicorn.conf.py and the GUNICORN_WORKERS=1 warning are unaffected.

Testing

tests/unit/test_mcp_service.py - 12 cases

class TestDiscover:
    def test_loads_builtin_services_in_order()                  # BUILTIN_SERVICES path -> SERVICE list, preserves BUILTIN_SERVICES order
    def test_skips_broken_service_if_not_requested(caplog)      # BUILTIN has a bad path, env doesn't mention -> WARN, no raise
    def test_raises_on_broken_service_if_requested()            # BUILTIN has a bad path, env mentions it -> RuntimeError
    def test_raises_when_service_module_lacks_SERVICE()         # Module exists but no SERVICE attribute
    def test_raises_when_SERVICE_is_wrong_type()                # SERVICE = "not a McpService"
    def test_raises_on_duplicate_service_names()                # Two paths export same name -> raises
    def test_raises_on_duplicate_mount_paths()                  # Two paths export same mount_path -> raises

class TestFilter:
    def test_unset_env_returns_all()                            # monkeypatch.delenv; all BUILTIN services present
    def test_empty_env_returns_zero()                           # MCP_SERVICES=''  -> []
    def test_whitelist_filters_by_name()                        # MCP_SERVICES='files' -> only files
    def test_whitelist_preserves_bulitin_order()                # MCP_SERVICES='files,spark_executor' -> order is BUILTIN, not env

class TestEnvErrors:
    def test_unknown_name_in_env_raises_with_available_list()   # MCP_SERVICES='bogus' -> RuntimeError mentions available names

tests/integration/test_main_lifespan.py - 5 cases

class TestLifespanMounting:
    def test_default_mounts_both_services()                     # no env -> /spark-executor-mcp and /files-mcp both reachable
    def test_MCP_SERVICES_files_only_mounts_one()              # /files-mcp reachable, /spark-executor-mcp 404
    def test_MCP_SERVICES_empty_mounts_none()                  # both 404 (only /health on root)
    def test_MCP_SERVICES_unknown_raises_at_startup()          # bad env -> TestClient startup raises
    def test_health_endpoint_unaffected_by_mcp_filtering()     # /health always 200

(Integration test uses TestClient(app) from main:app and exercises the real lifespan. monkeypatch is used to set/unset the env var per test.)

Existing tests (327)

• All pass unchanged.
• tests/integration/test_files_mcp_routes.py still works (it uses files_mcp.server:app directly, not the registry).
• tests/integration/test_mcp_routes.py for spark_executor still works the same way.

Risks

1. service.py import triggers the package's __init__. Each service.py does from <pkg> import app, which runs the package's __init__.py. For files_mcp, that means _init_root() runs (and may raise RuntimeError if FILES_MCP_ROOT is unset).

   • Mitigation: the existing files_mcp/__init__.py keeps its first try/except (tolerating unset FILES_MCP_ROOT at import time) - this was already documented and approved in the file-MCP spec. The conftest autouse fixture sets FILES_MCP_ROOT before tests run.
   • In production: operators must set FILES_MCP_ROOT before starting the server, as documented in the file-MCP spec.

2. Service-name / package-name coupling. The convention is "service name == package name" so the env whitelist is intuitive. A package that declares a different name will still work (the env var is matched by McpService.name, not by package), but the convention is the documentation hint.

   • Mitigation: the MCP_SERVICES env error message lists available service names explicitly.

3. Mount path collision across services. Two services with the same mount_path would cause fastapi-mcp to error at the second mount_http.

   • Mitigation: discover_and_filter checks for duplicate mount paths and raises a clear error before any mount happens.

4. Behavioral change at the wiring layer. Operators who read main.py looking for the service list will need to know about BUILTIN_SERVICES in common/mcp_service.py and the per-package service.py files.

   • Mitigation: the comment block in main.py (currently explaining the registry) is updated to point to common.mcp_service.BUILTIN_SERVICES and the manifest convention.

Implementation Order (preview - full plan comes from writing-plans)

1. common/mcp_service.py (dataclass + discover + filter) + tests/unit/test_mcp_service.py
2. spark_executor/service.py + files_mcp/service.py
3. Refactor main.py (remove old list + dataclass, use discover_and_filter)
4. tests/integration/test_main_lifespan.py
5. Run full suite; commit.

Acceptance Criteria

• uv run pytest -v passes (327 existing + 17 new = 344).
• git diff main --stat shows changes only in: common/mcp_service.py (new), spark_executor/service.py (new), files_mcp/service.py (new), main.py (modified), and 2 new test files.
• pyproject.toml / uv.lock unchanged.
• spark_executor/server.py, spark_executor/tools/*, spark_executor/core/*, files_mcp/server.py, files_mcp/tools/*, files_mcp/core/*, files_mcp/models.py, files_mcp/__init__.py are byte-for-byte unchanged.
• Smoke test: with no env var, both /spark-executor-mcp and /files-mcp mount successfully (same as today).
• Smoke test: with MCP_SERVICES=files, only /files-mcp mounts; /spark-executor-mcp returns 404.
• Smoke test: with MCP_SERVICES=files,bogus, startup raises with an error that lists Available: ['files', 'spark_executor'].