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

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

```python
# 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)."""
    raw = os.environ.get("MCP_SERVICES")
    if not raw:
        return services
    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:

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

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

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

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

```python
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']`.