server.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/mcp/server.py
 
"""
Purpose
-------
MCP server core (Layers 1 + orchestration) for the ``oct-mcp`` server.
Creates a :class:`mcp.server.fastmcp.FastMCP` instance, binds it to a
validated project root (Layer 1 — Gateway), and exposes the nine Phase
5A read-only tools via :mod:`oct.mcp.tools`.
 
Responsibilities
----------------
- Implement :func:`start_server`: resolve + validate project root,
  load config, instantiate all pipeline layers, register tools, run.
- Implement the token-bucket rate limiter (Layer 1) via
  :class:`RateLimiter`.
- Expose ``_server_state`` for tools to access the shared pipeline
  instances (singleton pattern: one server = one context).
 
Diagnostics
-----------
Domain: MCP
Levels:
    L1 — errors: startup failures, workspace binding failures
    L2 — lifecycle: server start/stop, workspace bound
    L3 — request details
    L4 — deep trace
 
Contracts
---------
- :func:`start_server` raises ``SystemExit(1)`` on fatal startup errors
  (invalid project root, no ``oct`` command) so ``__main__.py`` can
  propagate a clean exit code.
- The rate limiter is per-server-instance (not per-session), which is
  sufficient for the stdio Phase 5A transport where one server process
  handles one client session.
- ``_server_state`` is set before :meth:`FastMCP.run` is called and is
  available to all tool handlers during their execution.
"""
 
from __future__ import annotations
 
import os
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
 
try:
    from mcp.server.fastmcp import FastMCP
except ImportError as _mcp_err:  # pragma: no cover
    raise ImportError(
        "mcp>=1.0 is required for oct-mcp. "
        "Install with: pip install 'oct[mcp]'"
    ) from _mcp_err
 
try:
    from oc_diagnostics import _dbg as _real_dbg
 
    def _dbg(*args, **kwargs) -> None:
        _real_dbg(*args, **kwargs)
except ImportError:  # pragma: no cover
    def _dbg(*args, **kwargs) -> None:
        return None
 
from oct.mcp.audit import McpAuditLogger
from oct.mcp.config import McpConfig, load_config
from oct.mcp.executor import ToolExecutor
from oct.mcp.metrics import McpMetrics
from oct.mcp.policy import McpPolicy
from oct.mcp.redactor import Redactor
from oct.mcp.safety import SafetyGate
from oct.mcp.sandbox import SandboxExecutor, make_sandbox_backend
 
 
# ---------------------------------------------------------------------
# Rate limiter (Layer 1, token-bucket)
# ---------------------------------------------------------------------
 
 
class RateLimiter:
    """Simple token-bucket rate limiter (stdlib only).
 
    One token is consumed per tool call. Tokens refill at the rate of
    ``limit_per_minute / 60`` tokens per second. The bucket starts full.
 
    Thread safety: not needed for stdio Phase 5A (single-threaded).
    """
 
    def __init__(self, limit_per_minute: int = 30) -> None:
        self._limit = max(1, limit_per_minute)
        self._tokens: float = float(self._limit)
        self._last_refill: float = time.monotonic()
 
    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(
            float(self._limit),
            self._tokens + elapsed * (self._limit / 60.0),
        )
        self._last_refill = now
 
    def consume(self) -> bool:
        """Try to consume one token. Returns ``False`` if rate limit is exceeded."""
        self._refill()
        if self._tokens >= 1.0:
            self._tokens -= 1.0
            return True
        return False
 
 
# ---------------------------------------------------------------------
# Server state (shared pipeline instances)
# ---------------------------------------------------------------------
 
 
@dataclass
class ServerState:
    """All pipeline layer instances for one server session.
 
    Created in :func:`start_server` and stored in the module-level
    ``_server_state`` singleton so tool handlers can access it without
    having to pass context through the MCP SDK call graph.
    """
 
    project_root: Path
    config: McpConfig
    policy: McpPolicy
    executor: ToolExecutor
    redactor: Redactor
    audit_logger: McpAuditLogger
    rate_limiter: RateLimiter
    safety: SafetyGate
    metrics: McpMetrics
    session_id: str = ""
 
 
#: Module-level singleton. Set by :func:`start_server` before
#: :meth:`FastMCP.run` is called. ``None`` outside server context.
_server_state: ServerState | None = None
 
 
def get_server_state() -> ServerState:
    """Return the current :class:`ServerState`.
 
    Raises ``RuntimeError`` if called outside a running server context
    (i.e. before :func:`start_server` has initialised the state).
    """
    if _server_state is None:
        raise RuntimeError(
            "oct-mcp server state is not initialised. "
            "Call start_server() before accessing server state."
        )
    return _server_state
 
 
# ---------------------------------------------------------------------
# Workspace binding (Layer 1 — Gateway)
# ---------------------------------------------------------------------
 
 
def _resolve_project_root(root_arg: str | Path | None) -> Path:
    """Resolve and validate the project root directory.
 
    Parameters
    ----------
    root_arg
        Path provided via ``--project-root`` CLI argument, or ``None``
        to auto-discover from the current working directory.
 
    Returns
    -------
    Resolved absolute :class:`Path`.
 
    Raises
    ------
    SystemExit(1)
        If the directory does not exist or is not a directory.
    """
    func = "_resolve_project_root"
    if root_arg:
        candidate = Path(root_arg).resolve()
    else:
        try:
            from oct.core.project_root import find_project_root
            candidate = find_project_root(Path.cwd())
        except Exception:
            candidate = Path.cwd().resolve()
 
    if not candidate.exists():
        _dbg("MCP", func, f"SYSTEM_ERROR: project root not found: {candidate}", 1)
        print(f"oct-mcp: error: project root does not exist: {candidate}", file=sys.stderr)
        sys.exit(1)
 
    if not candidate.is_dir():
        _dbg("MCP", func, f"SYSTEM_ERROR: project root is not a directory: {candidate}", 1)
        print(
            f"oct-mcp: error: project root is not a directory: {candidate}",
            file=sys.stderr,
        )
        sys.exit(1)
 
    _dbg("MCP", func, f"project_root={candidate}", 2)
    return candidate
 
 
# ---------------------------------------------------------------------
# Server entry point
# ---------------------------------------------------------------------
 
 
def start_server(
    project_root: str | Path | None = None,
    profile: str | None = None,
    config_path: Path | None = None,
    transport: str = "stdio",
    host: str = "127.0.0.1",
    port: int = 3001,
    metrics_port: int | None = None,
) -> None:
    """Initialise and run the ``oct-mcp`` server.
 
    Steps:
 
    1. Resolve and validate project root (Layer 1 — Gateway).
    2. Load :class:`McpConfig` (with env var overrides).
    3. Apply centralised policy overrides (Phase 5C, M-C7).
    4. Override profile if ``--profile`` CLI arg was provided.
    5. Instantiate all pipeline layers.
    6. Start Prometheus metrics server (optional, Phase 5C).
    7. Create FastMCP server.
    8. Set ``_server_state`` singleton.
    9. Register 9 read-only (Phase 5A) + 11 write (Phase 5B) tools.
    10. Run the FastMCP server (stdio or SSE transport).
 
    Parameters
    ----------
    project_root
        Project root path override (``None`` = auto-discover).
    profile
        MCP profile override (``None`` = use config file / env var).
    config_path
        Path to the ``~/.oct-mcp/config.json`` config file
        (``None`` = use default location).
    transport
        MCP transport: ``"stdio"`` (default) or ``"sse"``
        (HTTP/Server-Sent Events for remote access).
    host
        Host to bind for SSE transport (default ``"127.0.0.1"``).
        Only used when *transport* is ``"sse"``.
    port
        Port to bind for SSE transport (default ``3001``).
        Only used when *transport* is ``"sse"``.
    metrics_port
        If set, starts a Prometheus ``/metrics`` server on this port.
        Overrides ``config.metrics_port``. Requires ``oct[metrics]``.
    """
    global _server_state
    func = "start_server"
 
    # Step 1: workspace binding
    root = _resolve_project_root(project_root)
 
    # Step 2: load config
    config = load_config(config_path)
 
    # Step 3: apply centralised policy overrides (Phase 5C)
    if config.policy_source:
        from oct.mcp.policy_loader import apply_policy_overrides, load_policy_source
        overrides = load_policy_source(config.policy_source)
        if overrides:
            config = apply_policy_overrides(config, overrides)  # type: ignore[assignment]
 
    # Step 4: profile override from CLI
    from oct.mcp.config import VALID_MCP_PROFILES
    if profile and profile in VALID_MCP_PROFILES:
        config.profile = profile
        _dbg("MCP", func, f"profile overridden to {profile} via CLI", 2)
 
    _dbg(
        "MCP", func,
        f"starting server profile={config.profile} transport={transport} root={root}",
        2,
    )
 
    # Step 5: instantiate pipeline layers
    import uuid
    session_id = str(uuid.uuid4())
 
    policy = McpPolicy(profile=config.profile, safe_mode=config.safe_mode)
    backend = make_sandbox_backend(config)
    sandbox = SandboxExecutor(
        backend=backend,
        memory_limit_mb=config.memory_limit_mb,
    )
    executor = ToolExecutor(
        sandbox=sandbox,
        timeout_default=config.timeout_default_seconds,
        timeout_test=config.timeout_test_seconds,
        max_output_bytes=config.max_output_bytes,
    )
    redactor = Redactor(
        redaction_patterns=_load_debug_config_patterns(root, config),
        max_output_bytes=config.max_output_bytes,
    )
    audit_logger = McpAuditLogger(
        audit_log_path=config.audit_log_path,
        max_files=config.audit_log_max_files,
        max_size_bytes=int(config.audit_log_max_size_mb * 1024 * 1024),
    )
    rate_limiter = RateLimiter(limit_per_minute=config.rate_limit_per_minute)
    safety = SafetyGate(config)
    metrics = McpMetrics()
 
    # Step 6: start Prometheus metrics server if enabled (Phase 5C)
    effective_metrics_port = metrics_port if metrics_port is not None else config.metrics_port
    if effective_metrics_port is not None:
        metrics.start_server(effective_metrics_port)
        _dbg("MCP", func, f"Prometheus metrics on port {effective_metrics_port}", 2)
 
    # Step 7: create FastMCP server and register tools
    mcp_server = FastMCP(
        name="oct-mcp",
        instructions=(
            "Option C Tools MCP server — exposes oct linter, health check, "
            "skeleton exporter, dependency checker, test runner, type checker, "
            "diagnostics, and git tools as MCP tools."
        ),
    )
 
    # Step 8: set server state singleton before registering tools
    _server_state = ServerState(
        project_root=root,
        config=config,
        policy=policy,
        executor=executor,
        redactor=redactor,
        audit_logger=audit_logger,
        rate_limiter=rate_limiter,
        safety=safety,
        metrics=metrics,
        session_id=session_id,
    )
 
    # Step 9: register tools (read-only Phase 5A + write Phase 5B)
    from oct.mcp import tools as _tools_module
    _tools_module.register_tools(mcp_server)
    _tools_module.register_write_tools(mcp_server)
 
    _dbg("MCP", func, f"server ready session_id={session_id}", 2)
 
    # Step 10: run (blocks until client disconnects)
    if transport == "sse":
        mcp_server.run(transport="sse", host=host, port=port)
    else:
        mcp_server.run(transport="stdio")
 
 
def _load_debug_config_patterns(root: Path, config: McpConfig) -> list:
    """Load ``redaction_patterns`` from ``debug_config.json`` if trusted.
 
    Only reads the project-level debug config when
    ``config.trust_repo_config`` is ``True`` and the profile is not
    ``"airgapped"`` or ``"strict"``.
 
    Returns an empty list (safe default) on any read/parse failure.
    """
    func = "_load_debug_config_patterns"
    if not config.trust_repo_config:
        return []
    if config.profile in ("strict", "airgapped"):
        return []
    # OI-504: canonical path since v0.9.0 is oc_diagnostics/debug_config.json.
    # Fall back to the legacy root-level path for backwards compatibility.
    canonical = root / "oc_diagnostics" / "debug_config.json"
    legacy = root / "debug_config.json"
    if canonical.exists():
        debug_cfg_path = canonical
    elif legacy.exists():
        _dbg(
            "MCP",
            func,
            f"using legacy debug_config.json at {legacy} — migrate to oc_diagnostics/",
            1,
        )
        debug_cfg_path = legacy
    else:
        return []
    try:
        import json
        raw = json.loads(debug_cfg_path.read_text(encoding="utf-8"))
        patterns = raw.get("redaction_patterns", [])
        if isinstance(patterns, list):
            _dbg("MCP", func, f"loaded {len(patterns)} redaction patterns", 2)
            return patterns
    except Exception as exc:
        _dbg("MCP", func, f"could not read debug_config.json: {exc}", 1)
    return []