acdl/mcp/atelier/server.py

"""mcp/atelier/server.py — Nova Atelier MCP server (REQ-223, D-135, D-137, D-140).

Plugin-registry architecture (D-140): plugins/<name>.py modules each expose
``register(mcp) -> None`` and call ``@mcp.tool()`` for their tools. This file
scans ``plugins/`` and calls ``register`` on each. Future capabilities drop
in as new plugin files — no server.py edits.

Transport: stdio (D-135). The MCP Python SDK v2 (``modelcontextprotocol/
python-sdk``, D-137) is the target. If the SDK is not installed, the server
degrades to a plain-Python tool registry that can be tested directly — the
tools are callable without MCP. This makes the server testable in CI
without the SDK installed.

Usage (with SDK):
    python3 -m mcp.atelier.server

Usage (without SDK, for testing):
    from mcp.atelier.server import NovaAtelierServer
    s = NovaAtelierServer()
    s.load_plugins()
    result = s.call_tool("atelier.lookup_principle", {"domain": "security", "principle_id": "P4"})
"""
from __future__ import annotations

import importlib
import json
import os
import pathlib
import sys
import types
from dataclasses import dataclass, field
from typing import Any, Callable

_PLUGIN_DIR = pathlib.Path(__file__).parent / "plugins"
_VENDOR_DIR = pathlib.Path(__file__).parent / "vendor"


class _ToolRegistry:
    """A minimal tool registry that mimics the MCP ``@mcp.tool()`` decorator.

    When the MCP SDK is available, ``NovaAtelierServer`` wraps a real
    ``MCPServer`` and the decorator registers tools with the SDK. When the
    SDK is absent, this registry is the fallback — tools are callable via
    ``call_tool()`` for testing.
    """

    def __init__(self) -> None:
        self._tools: dict[str, dict[str, Any]] = {}

    def tool(self, name: str | None = None, description: str | None = None) -> Callable:
        def decorator(fn: Callable) -> Callable:
            tool_name = name or fn.__name__
            self._tools[tool_name] = {
                "fn": fn,
                "description": description or fn.__doc__ or "",
                "name": tool_name,
            }
            return fn
        return decorator

    def list_tools(self) -> list[dict[str, str]]:
        return [{"name": t["name"], "description": t["description"]} for t in self._tools.values()]

    def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
        if name not in self._tools:
            raise KeyError(f"Unknown tool: {name}")
        return self._tools[name]["fn"](**arguments)


class NovaAtelierServer:
    """The Nova Atelier MCP server.

    Wraps an MCP SDK ``MCPServer`` if available; otherwise uses the
    ``_ToolRegistry`` fallback. Plugins are loaded from ``plugins/``.
    """

    def __init__(self) -> None:
        self.registry = _ToolRegistry()
        self._mcp = None
        try:
            from mcp.server import MCPServer  # type: ignore[import-not-found]
            self._mcp = MCPServer("atelier")
        except ImportError:
            pass  # SDK not installed — fallback to _ToolRegistry

    @property
    def mcp(self) -> Any:
        """The object plugins register tools on (real MCPServer or fallback)."""
        return self._mcp if self._mcp is not None else self.registry

    def load_plugins(self) -> list[str]:
        """Scan plugins/ and call ``register(mcp)`` on each. Returns loaded names."""
        loaded: list[str] = []
        for p in sorted(_PLUGIN_DIR.glob("*.py")):
            if p.stem == "__init__":
                continue
            mod_name = f"mcp.atelier.plugins.{p.stem}"
            mod = importlib.import_module(mod_name)
            if hasattr(mod, "register"):
                mod.register(self.mcp if self._mcp else self.registry)
                loaded.append(p.stem)
        return loaded

    def list_tools(self) -> list[dict[str, str]]:
        if self._mcp is not None:
            return [{"name": t.name, "description": t.description} for t in self._mcp._tools.values()]  # type: ignore[attr-defined]
        return self.registry.list_tools()

    def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
        if self._mcp is not None:
            raise RuntimeError("MCP SDK call_tool not supported in fallback mode — use the MCP client")
        return self.registry.call_tool(name, arguments)

    def run(self) -> None:
        """Run the server over stdio (requires the MCP SDK)."""
        if self._mcp is None:
            raise RuntimeError("MCP SDK not installed — cannot run server. Install: pip install mcp")
        self._mcp.run()


def _make_plugin_compat_decorator(registry_or_mcp: Any) -> Callable:
    """Return a ``tool()`` decorator that works for both the fallback
    registry and the real MCP SDK."""
    if hasattr(registry_or_mcp, "tool"):
        return registry_or_mcp.tool
    # Fallback: wrap registry.tool() as a decorator factory
    return registry_or_mcp.tool


def main() -> None:
    server = NovaAtelierServer()
    loaded = server.load_plugins()
    print(f"Atelier MCP server — {len(loaded)} plugins loaded: {', '.join(loaded)}", file=sys.stderr)
    if server._mcp is None:
        print("MCP SDK not installed — server is in fallback (test) mode.", file=sys.stderr)
        print("Tools: " + ", ".join(t["name"] for t in server.list_tools()), file=sys.stderr)
    else:
        server.run()


if __name__ == "__main__":
    main()