insights/tools/phrocs/mcp_server.py

#!/usr/bin/env python3
"""Local MCP server for Insights dev environment process observability.

Both tools query phrocs directly over a Unix domain socket whose path is
derived from the workspace root (CWD), matching the socket phrocs binds.
No file-based intermediary — phrocs tracks status, metrics, and logs in-memory.

Run via Claude Code's .mcp.json (invoked automatically by the MCP client):
  uv run python tools/phrocs/mcp_server.py
"""

from __future__ import annotations

import os
import re
import json
import socket
import hashlib
from typing import Any

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "phrocs",
    instructions=(
        "Tools for inspecting the local Insights dev environment. "
        "Start the dev environment with `./bin/start` before using these tools."
    ),
)


def _socket_path() -> str:
    """Return the phrocs socket path for the current workspace (CWD).

    Mirrors the SocketPathFor logic in tools/phrocs/internal/ipc/server.go so
    both sides agree on the path without any runtime coordination.
    """
    real = os.path.realpath(os.getcwd())
    digest = hashlib.sha256(real.encode()).hexdigest()[:8]
    return f"/tmp/phrocs-{digest}.sock"


_PHROCS_SOCK = _socket_path()


def _query_phrocs(cmd: dict) -> dict | None:
    """Send a command to phrocs via Unix domain socket.

    Returns the parsed response dict, or None if phrocs is not running.
    """
    try:
        with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
            sock.settimeout(2.0)
            sock.connect(_PHROCS_SOCK)
            sock.sendall((json.dumps(cmd) + "\n").encode())
            with sock.makefile() as f:
                return json.loads(f.readline())
    except (OSError, json.JSONDecodeError, TimeoutError):
        return None


_NOT_RUNNING = {"error": "phrocs is not running. Start the dev environment with: ./bin/start"}

# Log lines come from a VT emulator render: they can carry ANSI escapes and
# screen-width padding (the emulator hard-wraps at terminal width, so length
# is already bounded). Clean them so MCP clients don't pay tokens for noise.
_ANSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)")


def _clean_log_lines(lines: list[str]) -> list[str]:
    return _collapse_repeats([_ANSI_RE.sub("", line).rstrip() for line in lines])


def _collapse_repeats(lines: list[str]) -> list[str]:
    out: list[str] = []
    i = 0
    while i < len(lines):
        j = i + 1
        while j < len(lines) and lines[j] == lines[i]:
            j += 1
        out.append(lines[i])
        if j - i > 1:
            out.append(f"[repeated x{j - i}]")
        i = j
    return out


@mcp.tool()
def get_process_status(process: str = "") -> dict[str, Any]:
    """Get the status of one or all dev environment processes.
    Returns pid (OS process ID), running state, readiness, exit code,
    startup duration, and resource metrics (mem_rss_mb, cpu_percent,
    thread_count, etc.) for each process. Metrics arrive with ~5 seconds.
    Args:
        process: Process name (e.g. 'backend', 'frontend', 'celery-worker').
                 Leave empty to get status for all processes.
    """
    if process:
        result = _query_phrocs({"cmd": "status", "process": process})
        if result is None:
            return {process: _NOT_RUNNING}
        if not result.get("ok"):
            return {process: {"error": result.get("error", "unknown error")}}
        return {process: result}

    result = _query_phrocs({"cmd": "status_all"})
    if result is None:
        return _NOT_RUNNING
    if not result.get("ok"):
        return {"error": result.get("error", "unknown error")}
    return result.get("processes", {})


@mcp.tool()
def get_process_logs(process: str, lines: int = 100, grep: str = "") -> dict[str, Any]:
    """Get recent log output from a dev environment process.
    Reads from phrocs' in-memory scrollback buffer (10,000 lines per process).
    Prefer a grep pattern over a large line count when hunting for something
    specific — it filters server-side across the whole buffer.
    Args:
        process: Process name (e.g. 'backend', 'frontend', 'celery-worker').
        lines: Number of recent lines to return (default 100, max 500).
        grep: Optional regex pattern to filter lines (e.g. 'ERROR', 'warning').
    """
    lines = min(max(lines, 1), 500)
    result = _query_phrocs({"cmd": "logs", "process": process, "lines": lines, "grep": grep or ""})

    if result is None:
        return {"process": process, **_NOT_RUNNING}
    if not result.get("ok"):
        return {"process": process, "error": result.get("error", "unknown error")}

    resp: dict[str, Any] = {
        "process": process,
        "returned_lines": len(result["lines"]),
        "logs": "\n".join(_clean_log_lines(result["lines"])),
    }
    if grep:
        resp["grep"] = grep
        resp["total_matched"] = result.get("total_matched", 0)
    resp["buffered"] = result.get("buffered", 0)
    return resp


@mcp.tool()
def send_keys(process: str, keys: str) -> dict[str, Any]:
    """Send input to a dev environment process's stdin.
    Use this to answer interactive prompts, confirm dialogs, or send
    signals like Ctrl+C (as '\u0003') to a running process.
    Args:
        process: Process name (e.g. 'backend', 'frontend', 'celery-worker').
        keys: The text to send (e.g. 'yes\\n', 'Y\\n'). Include '\\n' for Enter.
    """
    result = _query_phrocs({"cmd": "send-keys", "process": process, "keys": keys})
    if result is None:
        return {"process": process, **_NOT_RUNNING}
    if not result.get("ok"):
        return {"process": process, "error": result.get("error", "unknown error")}
    return {"process": process, "ok": True}


@mcp.tool()
def toggle_process(process: str) -> dict[str, Any]:
    """Toggle a process: stop it if running, start it if stopped.
    Call this twice on a running process to restart it.
    Args:
        process: Process name to toggle.
    """
    result = _query_phrocs({"cmd": "toggle-proc", "process": process})
    if result is None:
        return {"process": process, **_NOT_RUNNING}
    if not result.get("ok"):
        return {"process": process, "error": result.get("error", "unknown error")}
    return {"process": process, "ok": True}


if __name__ == "__main__":
    mcp.run()