add homelab review skill

ai/skills/homelab-infrastructure-review/scripts/mcp_client.py

@@ -0,0 +1,115 @@
+#!/usr/bin/env python3
+"""
+MCP Homelable Integration Module.
+
+Communicates directly with the homelable-backend on port 9445
+for canvas topology queries, node/edge retrieval, and scanning.
+"""
+
+import os
+import json
+import logging
+import time
+from typing import Optional, List, Dict
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+BACKEND_URL = "http://localhost:9445/api/v1"
+ENV_PATH = "/srv/configs/docker_compose/homelabel/.env"
+
+
+def _get_service_key() -> str:
+    """Read service key dynamically from .env to authenticate request securely."""
+    try:
+        if os.path.exists(ENV_PATH):
+            with open(ENV_PATH, "r") as f:
+                for line in f:
+                    if line.startswith("MCP_SERVICE_KEY="):
+                        return line.split("=", 1)[1].strip()
+    except Exception as exc:
+        logger.error(f"Failed to read service key from {ENV_PATH}: {exc}")
+    return ""
+
+
+def health_check(endpoint: str = BACKEND_URL) -> bool:
+    """Check if homelable-backend endpoint is reachable."""
+    try:
+        import requests
+    except ImportError:
+        # Socket check fallback
+        import socket
+        try:
+            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            s.settimeout(3)
+            s.connect(('localhost', 9445))
+            s.close()
+            logger.info("Backend socket check: port 9445 is open")
+            return True
+        except (ConnectionRefusedError, socket.timeout, OSError):
+            logger.warning("Backend socket check failed: port 9445 unreachable")
+            return False
+
+    key = _get_service_key()
+    headers = {"X-MCP-Service-Key": key}
+    try:
+        resp = requests.get(f"{endpoint}/health", headers=headers, timeout=5)
+        if resp.status_code == 200:
+            logger.info("Homelable-backend health check: OK")
+            return True
+        else:
+            logger.warning(f"Homelable-backend health check failed with HTTP {resp.status_code}")
+            return False
+    except requests.RequestException as exc:
+        logger.warning(f"Homelable-backend health check exception: {exc}")
+        return False
+
+
+def fetch_nodes(endpoint: str = BACKEND_URL) -> Optional[List[Dict]]:
+    """Fetch nodes directly from homelable-backend."""
+    if not health_check(endpoint):
+        return None
+
+    try:
+        import requests
+    except ImportError:
+        logger.error("requests library not installed")
+        return None
+
+    key = _get_service_key()
+    headers = {"X-MCP-Service-Key": key}
+    try:
+        resp = requests.get(f"{endpoint}/nodes", headers=headers, timeout=5)
+        resp.raise_for_status()
+        nodes = resp.json()
+        logger.info(f"Fetched {len(nodes)} nodes directly from backend")
+        return nodes
+    except requests.RequestException as exc:
+        logger.error(f"Failed to fetch nodes from backend: {exc}")
+        return None
+
+
+def log_mcp_error(source: str, output: str) -> None:
+    """Write MCP-related errors to errors.log."""
+    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
+    log_line = f"[{timestamp}] ERROR: Homelable {source} {output}.\n"
+    errors_log_path = "/srv/configs/ai/errors.log"
+    try:
+        with open(errors_log_path, "a") as f:
+            f.write(log_line)
+    except OSError:
+        print(log_line, file=__import__("sys").stderr)
+
+
+if __name__ == "__main__":
+    logger.info("Testing backend integration...")
+    if health_check():
+        nodes = fetch_nodes()
+        if nodes:
+            print(f"\nSuccessfully retrieved {len(nodes)} nodes:")
+            for node in nodes[:5]:
+                print(f"  - {node.get('label', 'unknown')} ({node.get('ip', 'no IP')}) [Status: {node.get('status', 'n/a')}]")
+        else:
+            print("Failed to retrieve nodes.")
+    else:
+        print("Backend is unreachable.")