scripts/mcp_server.py - master - davydov/NGPowers

scripts/mcp_server.py
286 строк11 KB
Roman Davydov
feat(ngpowers): synthesize the ultimate autonomous developer ecosystem
09 авг 2026, 12:39
09 авг 2026, 12:3978901b8
#!/usr/bin/env python3
import os
import sys
import subprocess

try:
    from mcp.server.fastmcp import FastMCP
except ImportError:
    print("\n[NGPowers MCP] ERROR: The 'mcp' SDK is not installed.", file=sys.stderr)
    print("[NGPowers MCP] To run this server, please install it: pip install mcp\n", file=sys.stderr)
    sys.exit(1)

# Initialize FastMCP Server
mcp = FastMCP("ngpowers-triage")

script_dir = os.path.dirname(os.path.abspath(__file__))
ngpowers_root = os.path.abspath(os.path.join(script_dir, ".."))

def resolve_script_path(rel_path):
    normalized = os.path.normpath(rel_path)
    filename = os.path.basename(normalized)
    parts = normalized.split(os.sep)

    # Strip leading .ngpowers or .agents if present for fallback lookup
    if parts and parts[0] in (".ngpowers", ".agents"):
        stripped_rel = os.path.join(*parts[1:])
    else:
        stripped_rel = normalized

    candidates = [
        os.path.abspath(os.path.join(ngpowers_root, normalized)),
        os.path.abspath(os.path.join(ngpowers_root, stripped_rel)),
        os.path.abspath(os.path.join(os.getcwd(), normalized)),
        os.path.abspath(os.path.join(os.getcwd(), ".ngpowers", stripped_rel)),
        os.path.abspath(os.path.join(os.getcwd(), ".agents", stripped_rel)),
    ]

    # Search in skills/*/scripts/<filename>
    skills_dir = os.path.join(ngpowers_root, "skills")
    if os.path.isdir(skills_dir):
        for sk in os.listdir(skills_dir):
            sk_script = os.path.join(skills_dir, sk, "scripts", filename)
            candidates.append(sk_script)

    for candidate in candidates:
        if os.path.isfile(candidate):
            return candidate

    return candidates[0]

def run_script(script_path, args=[]):
    full_path = resolve_script_path(script_path)
    if not os.path.exists(full_path):
        return f"Error: Script not found at {script_path} (resolved: {full_path})"
    
    cmd = [sys.executable, full_path] + args
    res = subprocess.run(cmd, capture_output=True, text=True)
    if res.returncode != 0:
        return f"Execution Error:\nStdout: {res.stdout}\nStderr: {res.stderr}"
    return res.stdout

@mcp.tool()
def triage_go(command: str, package: str = "./...") -> str:
    """
    Triage Go compilation, linting, or tests.
    :param command: Either 'build', 'lint', or 'test'.
    :param package: Go package target.
    """
    return run_script(".ngpowers/skills/ngpowers-go/scripts/triage_go.py", [command, package])

@mcp.tool()
def triage_frontend(command: str) -> str:
    """
    Triage Next.js/React frontend compilation or linting.
    :param command: Either 'build' or 'lint'.
    """
    return run_script(".ngpowers/skills/ngpowers-frontend/scripts/triage_frontend.py", [command])

@mcp.tool()
def triage_security(command: str) -> str:
    """
    Triage security audits (gosec, npm-audit, trivy).
    :param command: Either 'gosec', 'npm-audit', or 'trivy'.
    """
    return run_script(".ngpowers/skills/ngpowers-go-security-scanner/scripts/triage_security.py", [command])

@mcp.tool()
def triage_qa(command: str) -> str:
    """
    Triage QA test runner (playwright, go-test).
    :param command: Either 'playwright' or 'go-test'.
    """
    return run_script(".ngpowers/skills/ngpowers-webapp-testing/scripts/triage_qa.py", [command])

@mcp.tool()
def manage_worktree(command: str, branch: str = "") -> str:
    """
    Manage isolated workspaces using Git Worktree.
    :param command: Either 'add', 'remove', 'list', 'prune', 'spawn', 'dashboard', or 'overlaps'.
    :param branch: Branch name for workspace.
    """
    args = [command]
    if branch:
        args.append(branch)
    return run_script("scripts/manage_worktree.py", args)

@mcp.tool()
def ngpowers_worktree_dashboard(as_json: bool = False) -> str:
    """
    Render visual status matrix of all active NGPowers Git Worktrees and SDLC phases.
    :param as_json: Set to true for JSON output format.
    """
    args = ["dashboard"]
    if as_json:
        args.append("--json")
    return run_script("scripts/manage_worktree.py", args)

@mcp.tool()
def ngpowers_check_worktree_conflicts(as_json: bool = False) -> str:
    """
    Check for overlapping file modifications across active parallel Git Worktrees.
    :param as_json: Set to true for JSON output format.
    """
    args = ["overlaps"]
    if as_json:
        args.append("--json")
    return run_script("scripts/manage_worktree.py", args)

@mcp.tool()
def ngpowers_spawn_worktree(branch: str, task_id: str = "", title: str = "") -> str:
    """
    Programmatically spawn an isolated Git Worktree workspace and initialize SDLC task state.
    :param branch: Target branch name to create.
    :param task_id: Optional task ID (e.g., TASK-105).
    :param title: Optional task title.
    """
    args = ["spawn", branch]
    if task_id:
        args.extend(["--task-id", task_id])
    if title:
        args.extend(["--title", title])
    return run_script("scripts/manage_worktree.py", args)

@mcp.tool()
def ngpowers_merge_worktree(branch: str, force: bool = False) -> str:
    """
    Full lifecycle merge: Validate Phase 4 -> Git Merge into target branch -> Archive Task -> Teardown Worktree.
    :param branch: Branch name to merge.
    :param force: Set to true to bypass verification quality gate.
    """
    args = ["merge", branch]
    if force:
        args.append("--force")
    return run_script("scripts/manage_worktree.py", args)

@mcp.tool()
def ngpowers_reverse_spec(target_dir: str = ".", spec_type: str = "service") -> str:
    """
    Reverse engineer brownfield codebase and generate baseline service_spec.md or frontend_spec.md.
    :param target_dir: Directory path to scan (default '.').
    :param spec_type: Either 'service' or 'frontend'.
    """
    return run_script("scripts/reverse_spec_gen.py", ["--dir", target_dir, "--type", spec_type, "--json"])

@mcp.tool()
def ngpowers_check_refactor_impact(symbol: str) -> str:
    """
    Analyze blast radius and occurrences of a symbol/struct/function across codebase before refactoring.
    :param symbol: Symbol name to analyze.
    """
    return run_script("scripts/check_refactor_impact.py", ["--symbol", symbol, "--json"])

@mcp.tool()
def ngpowers_get_phase_context(task_id: str, phase: int) -> str:
    """
    Get wave-based minimal context payload tailored for a specific SDLC phase (0..4) for token efficiency.
    :param task_id: Task ID.
    :param phase: Phase number (0 through 4).
    """
    return run_script("scripts/context_slice.py", ["--task", task_id, "--phase", str(phase), "--json"])

@mcp.tool()
def ngpowers_compact_code(file_path: str) -> str:
    """
    AST code summarizer: strips internal function bodies, retaining exported structs, interfaces, and signatures.
    :param file_path: Source code file path.
    """
    return run_script("scripts/compact_code_context.py", ["--file", file_path, "--json"])

@mcp.tool()
def ngpowers_memory_bank(action: str = "read", target: str = "activeContext", task_id: str = "", focus: str = "", files: str = "", phase: int = 0) -> str:
    """
    Persistent Memory Bank subsystem: resume subagent context instantly with ~150 tokens instead of reading chat history.
    :param action: Either 'read', 'update', or 'init'.
    :param target: Memory file: 'activeContext', 'progress', or 'systemPatterns'.
    :param task_id: Required for 'update' action.
    :param focus: Required for 'update' action.
    :param files: Target files string for 'update'.
    :param phase: Current SDLC phase number.
    """
    if action == "init":
        return run_script("scripts/memory_bank.py", ["init"])
    elif action == "update":
        args = ["update", "--task", task_id, "--focus", focus, "--phase", str(phase), "--json"]
        if files:
            args.extend(["--files", files])
        return run_script("scripts/memory_bank.py", args)
    else:
        return run_script("scripts/memory_bank.py", ["read", target, "--json"])

@mcp.tool()
def ngpowers_detect_spec_drift() -> str:
    """
    Architectural Spec Drift Detector (GRACE / MoAI-ADK standard): detects code modifications outpacing service_spec.md updates.
    """
    return run_script("scripts/detect_spec_drift.py", ["--json"])

@mcp.tool()
def ngpowers_fast_track(desc: str, file: str = "", task_id: str = "") -> str:
    """
    Fast Track Micro-Change Protocol: executes small single-line/config modifications atomically without 5-phase PRD overhead.
    :param desc: Short description of micro-change.
    :param file: Optional target file path.
    :param task_id: Optional task ID.
    """
    args = ["--desc", desc, "--json"]
    if file:
        args.extend(["--file", file])
    if task_id:
        args.extend(["--task", task_id])
    return run_script("scripts/fast_track_change.py", args)

@mcp.tool()
def ngpowers_validate_api_contract() -> str:
    """
    OpenAPI & Protobuf Contract Validator (Add REST API Endpoint 5⭐ Standard): checks Swagger/Proto contracts against code routes.
    """
    return run_script("scripts/validate_api_contract.py", ["--json"])

@mcp.tool()
def ngpowers_hotfix(desc: str, root_cause: str = "", resolution: str = "", task_id: str = "") -> str:
    """
    Config-Driven Emergency Bugfix Protocol & Auto-Postmortem (Fix Production Bug 5⭐ Standard).
    :param desc: Description of emergency bugfix.
    :param root_cause: Root cause explanation.
    :param resolution: Resolution description.
    :param task_id: Optional task ID.
    """
    args = ["--desc", desc, "--json"]
    if root_cause:
        args.extend(["--root-cause", root_cause])
    if resolution:
        args.extend(["--resolution", resolution])
    if task_id:
        args.extend(["--task", task_id])
    return run_script("scripts/hotfix_gate.py", args)

@mcp.tool()
def ngpowers_enforce_test_coverage() -> str:
    """
    TDD & Code Coverage Enforcement Gate: measures unit test coverage and flags untested code files.
    """
    return run_script("scripts/enforce_test_coverage.py", ["--json"])

@mcp.tool()
def ngpowers_security_owasp_scan() -> str:
    """
    OWASP Security & Secret Leak Scanner: scans for committed secret keys, tokens, SQL injections, and OWASP Top 10 risks.
    """
    return run_script("scripts/security_owasp_scan.py", ["--json"])

@mcp.tool()
def ngpowers_self_heal_loop(cmd: str = "", attempts: int = 3) -> str:
    """
    Self-Healing Auto-Fix Loop: captures failure tracebacks and applies targeted fix retries up to max attempts.
    :param cmd: Test or compile verification command.
    :param attempts: Maximum retry attempts (default 3).
    """
    args = ["--attempts", str(attempts), "--json"]
    if cmd:
        args.extend(["--cmd", cmd])
    return run_script("scripts/self_heal_loop.py", args)

if __name__ == "__main__":
    # Start FastMCP server (stdin/stdout transport by default)
    mcp.run()