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.MCPServer` instance (MCP Python SDK v2),
binds it to a validated project root (Layer 1 - Gateway), and exposes
the twenty Phase 5A/5B tools via :mod:`oct.mcp.tools`.
 
Responsibilities
----------------
- Implement :func:`build_mcp_server`: construct the SDK server object
  and register all twenty tools, without running any transport.
- 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, standard output that
  cannot be handed back to the stdio transport) so ``__main__.py`` can
  propagate a clean exit code.
- The SDK is MCP Python SDK v2 (``mcp>=2.2,<3``) with no v1 fallback:
  the two lines differ in threading and error semantics, and a security
  layer must not straddle them (plan 244, P1).
- Under v2 a synchronous tool handler runs on a worker thread
  (``anyio.to_thread.run_sync``) and concurrent requests overlap, so
  every tool call is serialised by the process-wide lock held in
  :mod:`oct.mcp.tools`. The rate limiter, executor, audit logger and
  metrics behind ``_server_state`` therefore keep the single-threaded
  contract they were written for (plan 244, P3).
- Under the ``stdio`` transport, standard output (fd 1) is the JSON-RPC
  channel. :func:`oct.mcp.__main__.main` diverts it to standard error
  for the startup window, and :func:`start_server` hands it back with
  :func:`oct.mcp.stdout_diversion.restore_stdout` immediately before
  ``MCPServer.run(transport="stdio")``, whose ``stdio_server()`` then
  diverts fd 1 again for the serving window. When the hand-back fails,
  :func:`start_server` exits with code 1 instead of calling ``run``,
  because the SDK would take the still-diverted fd 1 as its wire and
  serve the protocol on standard error. Nothing is written after
  ``run`` returns; ``_dbg`` flushes every record, so only an unflushed
  ``print()`` made while serving, or a record that a broken standard
  error rejected while serving, could still reach the wire when the
  SDK hands fd 1 back at shutdown.
- The rate limiter is per-server-instance (not per-session), which is
  sufficient for the stdio transport where one server process handles
  one client session.
- ``_server_state`` is set before :meth:`MCPServer.run` is called and is
  available to all tool handlers during their execution.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- oct.mcp.audit (McpAuditLogger, the Layer 6 audit log)
- oct.mcp.config (McpConfig, load_config and VALID_MCP_PROFILES for the server settings)
- oct.mcp.executor (ToolExecutor, which maps tool arguments to oct CLI calls)
- oct.mcp.metrics (McpMetrics, the optional Prometheus counters and exporter)
- oct.mcp.policy (McpPolicy, the Layer 3 profile rules)
- oct.mcp.redactor (Redactor, the Layer 5 output sanitiser)
- oct.mcp.safety (SafetyGate, the Layer 2.5 consent and dry-run gate)
- oct.mcp.sandbox (SandboxExecutor and make_sandbox_backend, the Layer 4 sandbox)
- oct.mcp.stdout_diversion (restore_stdout and stdout_is_diverted, which hand fd 1 back to the stdio transport and confirm it)
- oct.core.project_root (lazy; find_project_root when no project root is given)
- oct.mcp.policy_loader (lazy; the centralised policy overrides when a policy source is set)
- oct.mcp.tools (lazy; registers the twenty tools on the SDK server)
- mcp (MCPServer from the MCP Python SDK v2, and its transports)
"""
 
from __future__ import annotations
 
import os
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
 
from mcp.server import MCPServer
 
from oct.core.diagnostics import _dbg
 
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
from oct.mcp.stdout_diversion import restore_stdout, stdout_is_diverted
 
 
# ---------------------------------------------------------------------
# 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: this class is NOT internally synchronised. Under MCP
    SDK v2 a synchronous tool handler runs on a worker thread, so the
    bucket is protected instead by the process-wide tool-call lock in
    :mod:`oct.mcp.tools`, which serialises every call (plan 244, P3).
    """
 
    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.
 
    Thread safety: none of these instances is internally synchronised.
    MCP SDK v2 dispatches synchronous tool handlers to worker threads
    and lets requests overlap, so tool calls are serialised by the
    process-wide lock in :mod:`oct.mcp.tools`; only one call is ever
    inside the pipeline at a time (plan 244, P3).
    """
 
    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:`MCPServer.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
 
 
# ---------------------------------------------------------------------
# SDK server construction
# ---------------------------------------------------------------------
 
#: Instructions advertised to the AI host during MCP initialisation.
SERVER_INSTRUCTIONS: str = (
    "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."
)
 
 
def _oct_version() -> str:
    """Return the installed ``oct`` distribution version.
 
    The MCP server advertises this to the host as its ``version``. It is
    read from distribution metadata rather than written as a literal so
    the wire value can never drift from the installed package.
 
    Returns
    -------
    The ``oct`` version string, or ``"0.0.0"`` when ``oct`` is not
    installed as a distribution (a source checkout on ``sys.path``).
    """
    func = "_oct_version"
    try:
        from importlib.metadata import version as _dist_version
        return _dist_version("oct")
    except Exception as exc:  # pragma: no cover - source-checkout fallback
        _dbg("MCP", func, f"oct distribution metadata unavailable: {exc}", 1)
        return "0.0.0"
 
 
def build_mcp_server() -> MCPServer:
    """Create the SDK server object and register all twenty tools.
 
    Factored out of :func:`start_server` so the construction path can be
    exercised - and the advertised tool list inspected - without binding
    a transport (plan 244, W1).
 
    Returns
    -------
    An :class:`mcp.server.MCPServer` with the nine Phase 5A read-only
    tools and the eleven Phase 5B write tools registered.
    """
    func = "build_mcp_server"
    mcp_server = MCPServer(
        name="oct-mcp",
        instructions=SERVER_INSTRUCTIONS,
        version=_oct_version(),
    )
 
    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"registered tools version={mcp_server.version}", 2)
    return mcp_server
 
 
# ---------------------------------------------------------------------
# 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. Set ``_server_state`` singleton.
    8. Build the SDK server with all 20 tools registered
       (:func:`build_mcp_server`).
    9. Run the server (stdio, streamable HTTP, or SSE transport). Under
       stdio, first hand fd 1 back to the transport
       (:func:`oct.mcp.stdout_diversion.restore_stdout`; a no-op when no
       diversion is active, as in an in-process call), and exit with
       code 1 if a diversion is still recorded afterwards.
 
    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) for local AI hosts,
        ``"streamable-http"`` for remote access, or ``"sse"`` for the
        legacy HTTP/Server-Sent Events transport.
    host
        Host to bind for the HTTP transports (default ``"127.0.0.1"``).
        Ignored when *transport* is ``"stdio"``.
    port
        Port to bind for the HTTP transports (default ``3001``).
        Ignored when *transport* is ``"stdio"``.
    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: 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 8: build the SDK server with all 20 tools registered
    mcp_server = build_mcp_server()
 
    _dbg("MCP", func, f"server ready session_id={session_id}", 2)
 
    # Step 9: run (blocks until client disconnects)
    if transport == "streamable-http":
        mcp_server.run(transport="streamable-http", host=host, port=port)
    elif transport == "sse":
        mcp_server.run(transport="sse", host=host, port=port)
    else:
        # Under main(), every startup write above went to stderr. Hand fd 1
        # back to the wire; the SDK's stdio_server() diverts it again as soon
        # as run() enters the transport.
        restore_stdout()
        if stdout_is_diverted():
            # The hand-back failed, so fd 1 is still off the wire. The SDK
            # would take it as the wire and serve the protocol on stderr
            # (or nowhere), where the client never sees it; stop instead.
            _dbg(
                "MCP", func,
                "SYSTEM_ERROR: standard output is still diverted; "
                "the stdio transport was not started",
                1,
            )
            print(
                "oct-mcp: error: could not hand standard output back to the "
                "stdio transport",
                file=sys.stderr,
            )
            sys.exit(1)
        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 []