hive/core/verify_mcp.py

#!/usr/bin/env python3
"""
Verification script for Aden Hive Framework MCP Server

This script checks if the MCP server is properly installed and configured.
"""

import json
import logging
import subprocess
import sys
from pathlib import Path

logger = logging.getLogger(__name__)


def setup_logger():
    """Configure logger for CLI usage."""
    if not logger.handlers:
        handler = logging.StreamHandler(sys.stdout)
        formatter = logging.Formatter("%(message)s")
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        logger.setLevel(logging.INFO)


class Colors:
    GREEN = "\033[0;32m"
    YELLOW = "\033[1;33m"
    RED = "\033[0;31m"
    BLUE = "\033[0;34m"
    NC = "\033[0m"


def check(description: str) -> bool:
    """Print check description and return a context manager for result."""
    logger.info(f"Checking {description}... ", extra={"end": ""})
    sys.stdout.flush()
    return True


def success(msg: str = "OK"):
    """Log success message."""
    logger.info(f"{Colors.GREEN}✓ {msg}{Colors.NC}")


def warning(msg: str):
    """Log warning message."""
    logger.warning(f"{Colors.YELLOW}⚠ {msg}{Colors.NC}")


def error(msg: str):
    """Log error message."""
    logger.error(f"{Colors.RED}✗ {msg}{Colors.NC}")


def main():
    """Run verification checks."""
    setup_logger()
    logger.info("=== MCP Server Verification ===")
    logger.info("")

    script_dir = Path(__file__).parent.absolute()
    all_checks_passed = True

    # Check 1: Framework package installed
    check("framework package installation")
    try:
        result = subprocess.run(
            [sys.executable, "-c", "import framework; print(framework.__file__)"],
            capture_output=True,
            text=True,
            check=True,
            encoding="utf-8",
        )
        framework_path = result.stdout.strip()
        success(f"installed at {framework_path}")
    except subprocess.CalledProcessError:
        error("framework package not found")
        logger.info(f"  Run: uv pip install -e {script_dir}")
        all_checks_passed = False

    # Check 2: MCP dependencies
    check("MCP dependencies")
    missing_deps = []
    for dep in ["mcp", "fastmcp"]:
        try:
            subprocess.run(
                [sys.executable, "-c", f"import {dep}"],
                capture_output=True,
                check=True,
                encoding="utf-8",
            )
        except subprocess.CalledProcessError:
            missing_deps.append(dep)

    if missing_deps:
        error(f"missing: {', '.join(missing_deps)}")
        logger.info(f"  Run: uv pip install {' '.join(missing_deps)}")
        all_checks_passed = False
    else:
        success("all installed")

    # Check 3: MCP configuration file
    check("MCP configuration file")
    mcp_config = script_dir / ".mcp.json"
    if mcp_config.exists():
        try:
            with open(mcp_config, encoding="utf-8") as f:
                config = json.load(f)

            if "mcpServers" in config:
                success("found and valid")
                for name, server_config in config.get("mcpServers", {}).items():
                    logger.info(f"  Server: {name}")
                    logger.info(f"    Command: {server_config.get('command')}")
                    logger.info(f"    Args: {' '.join(server_config.get('args', []))}")
            else:
                warning("exists but missing mcpServers config")
                all_checks_passed = False
        except json.JSONDecodeError:
            error("invalid JSON format")
            all_checks_passed = False
    else:
        warning("not found (optional)")
        logger.info(f"  Location would be: {mcp_config}")

    # Check 4: Framework modules
    check("core framework modules")
    modules_to_check = [
        "framework.runtime.core",
        "framework.graph.executor",
        "framework.graph.node",
        "framework.builder.query",
        "framework.llm",
    ]

    failed_modules = []
    for module in modules_to_check:
        try:
            subprocess.run(
                [sys.executable, "-c", f"import {module}"],
                capture_output=True,
                check=True,
                encoding="utf-8",
            )
        except subprocess.CalledProcessError:
            failed_modules.append(module)

    if failed_modules:
        error(f"failed to import: {', '.join(failed_modules)}")
        all_checks_passed = False
    else:
        success(f"all {len(modules_to_check)} modules OK")

    logger.info("")
    logger.info("=" * 40)
    if all_checks_passed:
        logger.info(f"{Colors.GREEN}✓ All checks passed!{Colors.NC}")
        logger.info("")
        logger.info("Your framework is ready to use.")
    else:
        logger.info(f"{Colors.RED}✗ Some checks failed{Colors.NC}")
        logger.info("")
        logger.info("To fix issues, run:")
        logger.info(f"  uv run python {script_dir / 'setup_mcp.py'}")
    logger.info("")


if __name__ == "__main__":
    main()