""" Genesis n8n Local API Bridge A FastAPI server that n8n (on Elestio cloud) calls to execute local Genesis actions. WHY THIS EXISTS: n8n runs on Elestio cloud. Genesis scripts run on WSL local machine at /mnt/e/genesis-system. n8n's executeCommand nodes cannot reach WSL directly. This FastAPI server runs locally and bridges the gap: n8n calls HTTP → this runs the scripts. USAGE: # Start the server (add to tmux or systemd) python3 /mnt/e/genesis-system/core/n8n_local_api.py # Or with uvicorn for production uvicorn core.n8n_local_api:app --host 0.0.0.0 --port 5000 --reload n8n then calls: http://:5000/trigger-agent (WSL IP can be found via: hostname -I | awk '{print $1}') ENDPOINTS: POST /trigger-agent — Run a Genesis agent task via execution layer POST /update-dashboard — Update a dashboard with new data POST /health-check — Run genesis health check and return status POST /spawn-session — Spawn or resume a tmux session POST /run-script — Run a specific script by name (allowlisted) GET /health — Simple liveness check GET /status — Full system status """ import datetime import json import logging import os import subprocess import sys import time from pathlib import Path from typing import Any, Optional # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- GENESIS_ROOT = os.getenv("GENESIS_ROOT", "/mnt/e/genesis-system") API_PORT = int(os.getenv("N8N_LOCAL_API_PORT", "5000")) API_HOST = os.getenv("N8N_LOCAL_API_HOST", "0.0.0.0") # Security: only these scripts can be executed via /run-script ALLOWED_SCRIPTS: dict[str, str] = { "genesis-heartbeat": f"{GENESIS_ROOT}/core/genesis_heartbeat.py status", "heartbeat-pulse": f"{GENESIS_ROOT}/core/genesis_heartbeat.py pulse", "circuit-breaker": f"{GENESIS_ROOT}/core/circuit_breaker.py status", "memory-hub-health": f"{GENESIS_ROOT}/core/genesis_memory_hub.py --health", "youtube-scout": f"{GENESIS_ROOT}/scripts/youtube_to_genesis_memory.py", "memory-sync": f"{GENESIS_ROOT}/core/genesis_memory_hub.py", "rate-limit-status": f"{GENESIS_ROOT}/core/gemini_rate_maximizer.py status", } # Dashboard file locations DASHBOARD_PATHS: dict[str, str] = { "agent-status": f"{GENESIS_ROOT}/agent_status.txt", "hive-metrics": f"{GENESIS_ROOT}/hive/SWARM_METRICS_REPORT.md", } log = logging.getLogger("n8n_local_api") logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) # --------------------------------------------------------------------------- # FastAPI app (lazy import to avoid hard dependency at module level) # --------------------------------------------------------------------------- try: from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse import uvicorn _FASTAPI_AVAILABLE = True except ImportError: _FASTAPI_AVAILABLE = False def _run_command(cmd: str, timeout: int = 30) -> dict[str, Any]: """ Run a shell command safely and return structured output. Args: cmd: Shell command string timeout: Max seconds to wait Returns: Dict with stdout, stderr, returncode, success. """ try: result = subprocess.run( cmd, shell=True, capture_output=True, text=True, timeout=timeout, cwd=GENESIS_ROOT, ) return { "success": result.returncode == 0, "stdout": result.stdout.strip()[:2000], "stderr": result.stderr.strip()[:500], "returncode": result.returncode, } except subprocess.TimeoutExpired: return {"success": False, "error": f"Command timed out after {timeout}s", "returncode": -1} except Exception as e: return {"success": False, "error": str(e), "returncode": -1} def _get_system_status() -> dict[str, Any]: """Collect basic system health indicators.""" status: dict[str, Any] = { "timestamp": datetime.datetime.utcnow().isoformat(), "genesis_root_exists": Path(GENESIS_ROOT).exists(), "python_version": sys.version.split()[0], } # Check if tmux genesis session is running tmux = _run_command("tmux list-sessions 2>/dev/null || echo 'no tmux'") status["tmux_sessions"] = tmux.get("stdout", "unknown") # Check disk space on E: drive df = _run_command("df -h /mnt/e 2>/dev/null | tail -1") status["disk_e"] = df.get("stdout", "unknown") return status if _FASTAPI_AVAILABLE: app = FastAPI( title="Genesis n8n Local API Bridge", description=( "Bridges n8n (Elestio cloud) to Genesis scripts running on WSL local machine. " "n8n calls these endpoints instead of executeCommand nodes." ), version="1.0.0", ) # Allow n8n to call from Elestio domain app.add_middleware( CORSMiddleware, allow_origins=["*"], # Lock down in production allow_methods=["GET", "POST"], allow_headers=["*"], ) # ----------------------------------------------------------------------- # GET /health — simple liveness check for n8n HTTP Request nodes # ----------------------------------------------------------------------- @app.get("/health") async def health() -> dict[str, Any]: """ Liveness check. n8n calls this every 5 minutes to verify the bridge is alive. Returns: JSON: {"status": "ok", "timestamp": "...", "genesis_root": "..."} """ return { "status": "ok", "timestamp": datetime.datetime.utcnow().isoformat(), "genesis_root": GENESIS_ROOT, "genesis_root_accessible": Path(GENESIS_ROOT).exists(), } # ----------------------------------------------------------------------- # GET /status — full system status # ----------------------------------------------------------------------- @app.get("/status") async def status() -> dict[str, Any]: """ Full Genesis system status. Returns health of all key components. Returns: JSON with tmux sessions, disk space, Genesis root accessibility. """ return _get_system_status() # ----------------------------------------------------------------------- # POST /trigger-agent — dispatch a task to Genesis execution layer # ----------------------------------------------------------------------- @app.post("/trigger-agent") async def trigger_agent(request: Request) -> dict[str, Any]: """ Trigger a Genesis agent task. Called by n8n to dispatch work to the Genesis Execution Layer (Gemini swarm). Request body: { "task": "Summarise latest voice leads and send Telegram report", "priority": "high", // critical|high|normal|low "context": {} // optional extra context } Returns: JSON with execution result or queued task ID. """ try: body = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid JSON body") task = body.get("task", "").strip() priority = body.get("priority", "normal") context = body.get("context", {}) if not task: raise HTTPException(status_code=400, detail="'task' field is required") log.info(f"[trigger-agent] priority={priority} task={task[:80]}") # Try to use Genesis Execution Layer if available try: sys.path.insert(0, GENESIS_ROOT) from core.genesis_execution_layer import execute_task_sync result = execute_task_sync(task) return { "success": True, "task": task, "priority": priority, "result": str(result)[:1000], "timestamp": datetime.datetime.utcnow().isoformat(), } except ImportError: # Fallback: write to RWL queue for next agent pickup queue_path = Path(GENESIS_ROOT) / "loop" / "rwl_queue.json" try: if queue_path.exists(): with open(queue_path, "r") as f: queue = json.load(f) else: queue = [] queue.append({ "id": f"n8n-{int(time.time())}", "task": task, "priority": priority, "context": context, "source": "n8n_local_api", "queued_at": datetime.datetime.utcnow().isoformat(), "status": "pending", }) queue_path.parent.mkdir(parents=True, exist_ok=True) with open(queue_path, "w") as f: json.dump(queue, f, indent=2) return { "success": True, "task": task, "queued": True, "queue_path": str(queue_path), "timestamp": datetime.datetime.utcnow().isoformat(), } except Exception as e: return { "success": False, "task": task, "error": f"Execution layer unavailable and queue write failed: {e}", "timestamp": datetime.datetime.utcnow().isoformat(), } except Exception as e: return { "success": False, "task": task, "error": str(e), "timestamp": datetime.datetime.utcnow().isoformat(), } # ----------------------------------------------------------------------- # POST /update-dashboard — update a dashboard file with new data # ----------------------------------------------------------------------- @app.post("/update-dashboard") async def update_dashboard(request: Request) -> dict[str, Any]: """ Update a Genesis dashboard or status file with new data from n8n. Request body: { "dashboard": "agent-status", // key from DASHBOARD_PATHS "data": {"field": "value"}, // data to write "mode": "append" // append|replace (default: append) } Returns: JSON confirming write success. """ try: body = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid JSON body") dashboard = body.get("dashboard", "") data = body.get("data", {}) mode = body.get("mode", "append") if not dashboard: raise HTTPException(status_code=400, detail="'dashboard' field is required") path_str = DASHBOARD_PATHS.get(dashboard) if not path_str: raise HTTPException( status_code=400, detail=f"Unknown dashboard: {dashboard!r}. Known: {list(DASHBOARD_PATHS.keys())}", ) path = Path(path_str) timestamp = datetime.datetime.utcnow().isoformat() content = f"\n--- n8n update @ {timestamp} ---\n{json.dumps(data, indent=2)}\n" try: path.parent.mkdir(parents=True, exist_ok=True) if mode == "replace": path.write_text(content) else: with open(path, "a") as f: f.write(content) return { "success": True, "dashboard": dashboard, "path": str(path), "mode": mode, "timestamp": timestamp, } except Exception as e: return {"success": False, "dashboard": dashboard, "error": str(e)} # ----------------------------------------------------------------------- # POST /health-check — run Genesis health check scripts # ----------------------------------------------------------------------- @app.post("/health-check") @app.get("/health-check") async def run_health_check() -> dict[str, Any]: """ Run the Genesis health check and return structured results. Called by the Health Monitor n8n workflow instead of executeCommand. Returns: JSON with heartbeat and circuit breaker status. """ timestamp = datetime.datetime.utcnow().isoformat() # Check heartbeat heartbeat_cmd = f"python3 {GENESIS_ROOT}/core/genesis_heartbeat.py status 2>/dev/null" heartbeat = _run_command(heartbeat_cmd, timeout=10) try: heartbeat_data = json.loads(heartbeat.get("stdout", "{}")) except json.JSONDecodeError: heartbeat_data = {"raw": heartbeat.get("stdout", ""), "error": heartbeat.get("stderr", "")} # Check circuit breakers circuit_cmd = f"python3 {GENESIS_ROOT}/core/circuit_breaker.py status 2>/dev/null" circuit = _run_command(circuit_cmd, timeout=10) try: circuit_data = json.loads(circuit.get("stdout", "{}")) except json.JSONDecodeError: circuit_data = {"raw": circuit.get("stdout", ""), "error": circuit.get("stderr", "")} # Determine overall health heartbeat_ok = not heartbeat_data.get("error") and heartbeat.get("success", False) circuit_ok = not any( v.get("state") == "open" for v in (circuit_data.values() if isinstance(circuit_data, dict) else []) ) return { "timestamp": timestamp, "healthy": heartbeat_ok and circuit_ok, "heartbeat": heartbeat_data, "circuit_breakers": circuit_data, "genesis_root_accessible": Path(GENESIS_ROOT).exists(), } # ----------------------------------------------------------------------- # POST /spawn-session — spawn or resume tmux session # ----------------------------------------------------------------------- @app.post("/spawn-session") async def spawn_session(request: Request) -> dict[str, Any]: """ Spawn or resume a tmux session. Called by the Continuous Agent Spawner n8n workflow instead of executeCommand. Request body: { "session": "genesis", // tmux session name "command": "claude --continue" // command to run if session doesn't exist } Returns: JSON with session status. """ try: body = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid JSON body") session = body.get("session", "genesis") command = body.get("command", "claude --continue") # Check if session exists check = _run_command(f"tmux has-session -t {session} 2>/dev/null && echo running || echo stopped") session_state = check.get("stdout", "").strip() if session_state == "running": return { "success": True, "session": session, "action": "already_running", "timestamp": datetime.datetime.utcnow().isoformat(), } # Spawn new session spawn_cmd = f"cd {GENESIS_ROOT} && tmux new-session -d -s {session} '{command}'" spawn = _run_command(spawn_cmd, timeout=15) return { "success": spawn.get("success", False), "session": session, "action": "spawned" if spawn.get("success") else "failed", "error": spawn.get("stderr", "") or spawn.get("error", ""), "timestamp": datetime.datetime.utcnow().isoformat(), } # ----------------------------------------------------------------------- # POST /run-script — run an allowlisted script # ----------------------------------------------------------------------- @app.post("/run-script") async def run_script(request: Request) -> dict[str, Any]: """ Run a specific allowlisted Genesis script. Only scripts in ALLOWED_SCRIPTS can be executed. No arbitrary command execution. Request body: { "script": "youtube-scout", // key from ALLOWED_SCRIPTS "args": "--dry-run" // optional extra args } Returns: JSON with stdout/stderr/success. """ try: body = await request.json() except Exception: raise HTTPException(status_code=400, detail="Invalid JSON body") script_name = body.get("script", "") args = body.get("args", "") if script_name not in ALLOWED_SCRIPTS: raise HTTPException( status_code=400, detail=f"Script {script_name!r} not in allowlist. " f"Allowed: {list(ALLOWED_SCRIPTS.keys())}", ) base_cmd = ALLOWED_SCRIPTS[script_name] full_cmd = f"python3 {base_cmd} {args}".strip() if not base_cmd.startswith("python3") \ else f"{base_cmd} {args}".strip() log.info(f"[run-script] {script_name}: {full_cmd[:80]}") result = _run_command(full_cmd, timeout=60) return { "script": script_name, "success": result.get("success", False), "stdout": result.get("stdout", ""), "stderr": result.get("stderr", ""), "returncode": result.get("returncode", -1), "timestamp": datetime.datetime.utcnow().isoformat(), } # --------------------------------------------------------------------------- # CLI entrypoint # --------------------------------------------------------------------------- if __name__ == "__main__": if not _FASTAPI_AVAILABLE: print("ERROR: FastAPI and uvicorn are required.") print("Install with: pip3 install fastapi uvicorn --break-system-packages") sys.exit(1) # Print startup info print(f"\n=== Genesis n8n Local API Bridge ===") print(f"Listening on: http://{API_HOST}:{API_PORT}") print(f"Genesis root: {GENESIS_ROOT}") print(f"Genesis root accessible: {Path(GENESIS_ROOT).exists()}") print() print("Endpoints:") print(" GET /health — liveness check") print(" GET /status — full system status") print(" POST /trigger-agent — dispatch agent task") print(" POST /update-dashboard — update dashboard file") print(" POST /health-check — run health scripts") print(" POST /spawn-session — spawn tmux session") print(" POST /run-script — run allowlisted script") print() print("Allowed scripts:") for name, cmd in ALLOWED_SCRIPTS.items(): print(f" {name}: {cmd}") print() # Get WSL IP for n8n configuration ip_result = _run_command("hostname -I 2>/dev/null | awk '{print $1}'") wsl_ip = ip_result.get("stdout", "unknown").strip() print(f"WSL IP (use in n8n HTTP Request nodes): {wsl_ip}") print(f"n8n should call: http://{wsl_ip}:{API_PORT}/health-check") print() uvicorn.run( "core.n8n_local_api:app", host=API_HOST, port=API_PORT, reload=False, log_level="info", )