""" Genesis n8n Trigger Bridge Allows Claude Code and any Genesis agent to fire n8n webhooks programmatically. Usage: from core.n8n_trigger import trigger_webhook, notify_lead_captured # Fire a lead capture event result = notify_lead_captured("John Smith", "+61412345678", "concreting") # Fire any arbitrary webhook result = trigger_webhook("stripe-event", {"type": "checkout.session.completed"}) # Trigger a workflow by ID via the n8n management API result = trigger_workflow_by_id("abc123", {"key": "value"}) """ import datetime import os import requests from typing import Any, Optional # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- N8N_BASE_URL = os.getenv("N8N_BASE_URL", "https://n8n-genesis-u50607.vm.elestio.app") N8N_API_KEY = os.getenv("N8N_API_KEY", "") # Map of logical event names → webhook paths # These are the paths registered in n8n webhook nodes WEBHOOK_ENDPOINTS: dict[str, str] = { "telnyx-lead": "/webhook/telnyx-lead", "telnyx-lead-legacy":"/webhook/telnyx-lead-capture", "stripe-event": "/webhook/stripe-lifecycle", "health-check": "/webhook/genesis-health", "agent-complete": "/webhook/agent-completion", "youtube-scout": "/webhook/youtube-scout", "ghl-contact": "/webhook/ghl-contact-created", "ghl-pipeline": "/webhook/ghl-pipeline", "claude-dispatch": "/webhook/claude-dispatch", "instantly-reply": "/webhook/instantly-reply", "calcom-booking": "/webhook/calcom-booking", "telnyx-call": "/webhook/telnyx-call-events", "lead-nurture": "/webhook/lead-nurture-start", # Claude bridge webhooks (claude-bridge-workflow.json) "claude-desktop-to-code": "/webhook/claude-desktop-to-code", "claude-code-to-desktop": "/webhook/claude-code-to-desktop", } # --------------------------------------------------------------------------- # Core webhook trigger # --------------------------------------------------------------------------- def trigger_webhook( event_name: str, payload: dict[str, Any], timeout: int = 10, ) -> dict[str, Any]: """ Fire an n8n webhook by logical event name. Args: event_name: Key from WEBHOOK_ENDPOINTS (e.g. 'telnyx-lead') payload: Dict to POST as JSON body timeout: Request timeout in seconds Returns: Dict with 'status' (HTTP code) and 'response' (body) on success, or 'error' key on failure. """ endpoint = WEBHOOK_ENDPOINTS.get(event_name) if not endpoint: return { "error": f"Unknown event: {event_name!r}", "known": list(WEBHOOK_ENDPOINTS.keys()), } url = f"{N8N_BASE_URL}{endpoint}" try: resp = requests.post(url, json=payload, timeout=timeout) return { "status": resp.status_code, "response": resp.text[:500], "url": url, } except requests.exceptions.ConnectionError as e: return {"error": f"Connection failed to {url}: {e}"} except requests.exceptions.Timeout: return {"error": f"Timeout after {timeout}s calling {url}"} except Exception as e: return {"error": str(e)} def trigger_webhook_raw( path: str, payload: dict[str, Any], timeout: int = 10, ) -> dict[str, Any]: """ POST to an arbitrary n8n webhook path (no lookup table needed). Args: path: Webhook path, e.g. '/webhook/my-custom-path' payload: Dict to POST as JSON body timeout: Request timeout in seconds Returns: Dict with 'status' and 'response', or 'error'. """ url = f"{N8N_BASE_URL}{path}" try: resp = requests.post(url, json=payload, timeout=timeout) return {"status": resp.status_code, "response": resp.text[:500], "url": url} except Exception as e: return {"error": str(e)} # --------------------------------------------------------------------------- # Management API (requires N8N_API_KEY) # --------------------------------------------------------------------------- def trigger_workflow_by_id( workflow_id: str, payload: Optional[dict[str, Any]] = None, timeout: int = 15, ) -> dict[str, Any]: """ Trigger a specific n8n workflow execution via the management API. Requires N8N_API_KEY environment variable to be set. Args: workflow_id: The n8n workflow UUID or ID string payload: Optional data to pass as workflowData timeout: Request timeout in seconds Returns: Dict with execution data or error. """ if not N8N_API_KEY: return {"error": "N8N_API_KEY not set in environment"} url = f"{N8N_BASE_URL}/api/v1/workflows/{workflow_id}/run" headers = { "X-N8N-API-KEY": N8N_API_KEY, "Content-Type": "application/json", } try: resp = requests.post( url, headers=headers, json={"workflowData": payload or {}}, timeout=timeout, ) return {"status": resp.status_code, "response": resp.json()} except Exception as e: return {"error": str(e)} def list_workflows(timeout: int = 10) -> dict[str, Any]: """ List all workflows in the n8n instance via the management API. Requires N8N_API_KEY. Returns: Dict with 'workflows' list or 'error'. """ if not N8N_API_KEY: return {"error": "N8N_API_KEY not set in environment"} url = f"{N8N_BASE_URL}/api/v1/workflows" headers = {"X-N8N-API-KEY": N8N_API_KEY} try: resp = requests.get(url, headers=headers, timeout=timeout) if resp.status_code == 200: data = resp.json() return {"workflows": data.get("data", []), "count": len(data.get("data", []))} return {"error": f"HTTP {resp.status_code}: {resp.text[:200]}"} except Exception as e: return {"error": str(e)} def get_workflow_status(workflow_id: str, timeout: int = 10) -> dict[str, Any]: """ Get the current status of a specific workflow. Requires N8N_API_KEY. """ if not N8N_API_KEY: return {"error": "N8N_API_KEY not set in environment"} url = f"{N8N_BASE_URL}/api/v1/workflows/{workflow_id}" headers = {"X-N8N-API-KEY": N8N_API_KEY} try: resp = requests.get(url, headers=headers, timeout=timeout) if resp.status_code == 200: data = resp.json() return { "id": data.get("id"), "name": data.get("name"), "active": data.get("active"), "updatedAt": data.get("updatedAt"), } return {"error": f"HTTP {resp.status_code}: {resp.text[:200]}"} except Exception as e: return {"error": str(e)} # --------------------------------------------------------------------------- # High-level convenience functions — call these from agents # --------------------------------------------------------------------------- def notify_lead_captured( customer_name: str, phone: str, service: str, source: str = "telnyx", extra: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """ Fire lead capture webhook — call this after every voice conversation. This triggers the Telnyx Lead → GHL CRM workflow in n8n which: - Creates/updates contact in GHL - Tags with service interest - Triggers SMS + email follow-up - Stores in PostgreSQL leads table Args: customer_name: Full name of the lead phone: Phone number (E.164 format preferred, e.g. +61412345678) service: What service they're interested in (e.g. 'concreting') source: Lead source (default: 'telnyx') extra: Any additional fields to include in the payload Returns: Dict with webhook response or error. """ payload: dict[str, Any] = { "customer_name": customer_name, "phone_number": phone, "service_interest": service, "source": source, "timestamp": datetime.datetime.utcnow().isoformat(), } if extra: payload.update(extra) return trigger_webhook("telnyx-lead", payload) def notify_stripe_event( event_type: str, customer_email: str, business_name: str, plan: str, stripe_customer_id: str, extra: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """ Fire a Stripe lifecycle event to n8n. Triggers the Stripe Lifecycle Handler which provisions/offboards customers. Common event_types: - 'checkout.session.completed' → provision customer - 'invoice.payment_failed' → start dunning sequence - 'customer.subscription.deleted' → offboard Args: event_type: Stripe event type string customer_email: Customer's email address business_name: Their business name plan: Subscription plan ('starter'/'professional'/'agency') stripe_customer_id: Stripe cus_xxx identifier extra: Additional fields Returns: Dict with webhook response or error. """ payload: dict[str, Any] = { "type": event_type, "customer_email": customer_email, "business_name": business_name, "plan": plan, "stripe_customer_id": stripe_customer_id, "timestamp": datetime.datetime.utcnow().isoformat(), } if extra: payload.update(extra) return trigger_webhook("stripe-event", payload) def dispatch_agent_task( task: str, priority: str = "normal", agent_type: str = "gemini", context: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """ Dispatch a task to n8n for Gemini swarm execution. n8n will route this to the Genesis execution layer. Args: task: Natural language task description priority: 'critical', 'high', 'normal', or 'low' agent_type: 'gemini', 'claude', or 'openrouter' context: Optional additional context dict Returns: Dict with webhook response or error. """ payload: dict[str, Any] = { "task": task, "priority": priority, "agent_type": agent_type, "timestamp": datetime.datetime.utcnow().isoformat(), } if context: payload["context"] = context return trigger_webhook("claude-dispatch", payload) def start_lead_nurture_sequence( first_name: str, phone: str, email: str = "", service_interest: str = "AI receptionist", contact_id: str = "", extra: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """ Start the 5-touch MCTB nurture sequence for a new lead. Sequence (lead_nurture_mctb workflow): - T+0: SMS welcome - T+2h: Email with voice demo link - T+24h: SMS check-in - T+72h: Email case study (George's story) - T+7d: SMS final follow-up Args: first_name: Lead's first name phone: Phone number (E.164 preferred) email: Email address (optional but needed for email touches) service_interest: Their trade/service interest contact_id: GHL contact ID (if already created) extra: Additional fields Returns: Dict with webhook response or error. """ payload: dict[str, Any] = { "firstName": first_name, "phone": phone, "email": email, "service_interest": service_interest, "contact_id": contact_id, "timestamp": datetime.datetime.utcnow().isoformat(), } if extra: payload.update(extra) return trigger_webhook("lead-nurture", payload) def trigger_review_request( contact_id: str, contact_name: str, phone: str, email: str = "", opportunity_id: str = "", ) -> dict[str, Any]: """ Trigger the review request workflow after job completion. Sends SMS review request after 2 hours, follow-up email at 48 hours if no click detected. Args: contact_id: GHL contact ID contact_name: Full name for personalisation phone: Phone number for SMS email: Email for follow-up (optional) opportunity_id: GHL opportunity ID Returns: Dict with webhook response or error. """ payload: dict[str, Any] = { "contact_id": contact_id, "contact_name": contact_name, "phone": phone, "email": email, "opportunity_id": opportunity_id, "pipeline_stage_name": "Job Complete", "timestamp": datetime.datetime.utcnow().isoformat(), } return trigger_webhook("ghl-pipeline", payload) def health_ping() -> dict[str, Any]: """ Send a health ping to n8n genesis-health webhook. Use for confirming the webhook bridge is live. """ return trigger_webhook("health-check", { "source": "n8n_trigger.py", "timestamp": datetime.datetime.utcnow().isoformat(), "message": "Genesis health ping", }) # --------------------------------------------------------------------------- # Self-test # --------------------------------------------------------------------------- if __name__ == "__main__": import json import sys print("=== Genesis n8n Trigger Bridge — Self Test ===\n") print(f"N8N_BASE_URL: {N8N_BASE_URL}") print(f"N8N_API_KEY: {'SET (' + N8N_API_KEY[:8] + '...)' if N8N_API_KEY else 'NOT SET'}") print() # Test 1: Health ping via webhook print("[1] Health check webhook...") result = health_ping() print(f" Result: {json.dumps(result, indent=6)}") print() # Test 2: Simulate lead capture print("[2] Lead capture test (DRY RUN — sending to n8n)...") result = notify_lead_captured( customer_name="TEST LEAD", phone="+61400000000", service="test-service", source="n8n_trigger_selftest", ) print(f" Result: {json.dumps(result, indent=6)}") print() # Test 3: List workflows (requires API key) if N8N_API_KEY: print("[3] Listing workflows (requires N8N_API_KEY)...") result = list_workflows() if "workflows" in result: for wf in result["workflows"]: status = "ACTIVE" if wf.get("active") else "inactive" print(f" [{status}] {wf.get('name', 'unnamed')} (id={wf.get('id')})") else: print(f" Error: {result}") else: print("[3] Skipping workflow list — N8N_API_KEY not set") print(" Set N8N_API_KEY in config/secrets.env to enable management API calls") print("\n=== Self test complete ===") sys.exit(0)