116 lines
3.8 KiB
Python
116 lines
3.8 KiB
Python
#!/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.")
|