#!/usr/bin/env python3 """ ReceptionistAI - Lead Capture Webhook Handler ============================================== Receives leads from website forms and creates contacts in GHL. Run as a Flask server: python lead_capture_webhook.py --serve --port 5050 """ import os import sys import json import requests import argparse from datetime import datetime from pathlib import Path from flask import Flask, request, jsonify sys.path.insert(0, str(Path(__file__).parent.parent.parent)) app = Flask(__name__) class LeadCaptureHandler: """Handles incoming leads and syncs to GHL.""" def __init__(self): self.ghl_key = os.getenv("GHL_API_KEY", "pit-91962a1c-ae7f-4ef4-8335-2d61f27faf22") self.location_id = os.getenv("GHL_LOCATION_ID", "73q7bKDm2d6hsCtHuz1m") self.ghl_base = "https://services.leadconnectorhq.com" # Notification settings self.notify_email = os.getenv("NOTIFY_EMAIL", "kinan@agileadapt.com") self.slack_webhook = os.getenv("SLACK_WEBHOOK") def _ghl_request(self, method: str, endpoint: str, data: dict = None) -> dict: """Make request to GHL API.""" url = f"{self.ghl_base}{endpoint}" headers = { "Authorization": f"Bearer {self.ghl_key}", "Content-Type": "application/json", "Version": "2021-07-28" } try: if method == "POST": response = requests.post(url, headers=headers, json=data, timeout=30) elif method == "GET": response = requests.get(url, headers=headers, params=data, timeout=30) else: return {"error": f"Unsupported method: {method}"} if response.status_code in [200, 201]: return response.json() if response.text else {"status": "success"} else: return {"error": response.text, "status_code": response.status_code} except Exception as e: return {"error": str(e)} def process_lead(self, lead_data: dict) -> dict: """Process incoming lead and create GHL contact.""" # Extract and validate data name = lead_data.get("name", "").strip() email = lead_data.get("email", "").strip() phone = lead_data.get("phone", "").strip() business_name = lead_data.get("business_name", "").strip() business_type = lead_data.get("business_type", "").strip() message = lead_data.get("message", "").strip() source = lead_data.get("source", "website") if not email and not phone: return {"error": "Either email or phone is required"} # Calculate estimated lost revenue if data provided missed_calls = lead_data.get("missed_calls_week", 0) avg_job_value = lead_data.get("avg_job_value", 0) monthly_loss = missed_calls * avg_job_value * 4 if missed_calls and avg_job_value else 0 # Create GHL contact contact_data = { "locationId": self.location_id, "name": name or business_name or "Unknown", "email": email, "phone": phone, "companyName": business_name, "source": source, "tags": ["receptionistai", "lead", source], "customFields": [] } # Add custom fields if available if business_type: contact_data["customFields"].append({ "key": "business_type", "value": business_type }) if missed_calls: contact_data["customFields"].append({ "key": "missed_calls_week", "value": str(missed_calls) }) if avg_job_value: contact_data["customFields"].append({ "key": "avg_job_value", "value": str(avg_job_value) }) if monthly_loss: contact_data["customFields"].append({ "key": "monthly_loss", "value": str(monthly_loss) }) # Create contact in GHL result = self._ghl_request("POST", "/contacts/", contact_data) # Log the lead log_entry = { "timestamp": datetime.utcnow().isoformat(), "lead": { "name": name, "email": email, "phone": phone, "business": business_name, "type": business_type, "monthly_loss": monthly_loss }, "ghl_result": result, "source": source } # Append to log file log_path = Path(__file__).parent.parent / "logs" / "leads.jsonl" log_path.parent.mkdir(exist_ok=True) with open(log_path, "a") as f: f.write(json.dumps(log_entry) + "\n") # Send Slack notification if configured if self.slack_webhook: self._notify_slack(log_entry) return { "status": "success", "message": "Lead captured", "contact_id": result.get("contact", {}).get("id") if "contact" in result else None, "monthly_loss_estimate": f"${monthly_loss:,.0f}" if monthly_loss else None } def _notify_slack(self, lead_data: dict): """Send Slack notification for new lead.""" lead = lead_data.get("lead", {}) message = { "text": f"🎯 *New ReceptionistAI Lead!*\n\n" f"*Name:* {lead.get('name', 'N/A')}\n" f"*Business:* {lead.get('business', 'N/A')}\n" f"*Type:* {lead.get('type', 'N/A')}\n" f"*Email:* {lead.get('email', 'N/A')}\n" f"*Phone:* {lead.get('phone', 'N/A')}\n" f"*Est. Monthly Loss:* ${lead.get('monthly_loss', 0):,.0f}\n" f"*Source:* {lead_data.get('source', 'website')}" } try: requests.post(self.slack_webhook, json=message, timeout=10) except: pass # Flask routes handler = LeadCaptureHandler() @app.route("/webhook/lead", methods=["POST"]) def capture_lead(): """Webhook endpoint for lead capture.""" try: if request.is_json: data = request.get_json() else: data = request.form.to_dict() result = handler.process_lead(data) return jsonify(result), 200 if result.get("status") == "success" else 400 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/webhook/demo-request", methods=["POST"]) def demo_request(): """Webhook for demo call requests.""" try: if request.is_json: data = request.get_json() else: data = request.form.to_dict() # Add demo tag data["source"] = "demo_request" data["tags"] = ["demo_requested"] result = handler.process_lead(data) # TODO: Trigger demo call via VAPI # This would initiate an outbound call to the lead using Sarah demo agent return jsonify(result), 200 if result.get("status") == "success" else 400 except Exception as e: return jsonify({"error": str(e)}), 500 @app.route("/health", methods=["GET"]) def health(): """Health check endpoint.""" return jsonify({"status": "healthy", "service": "receptionistai-lead-capture"}) def main(): parser = argparse.ArgumentParser(description="Lead Capture Webhook for ReceptionistAI") parser.add_argument("--serve", action="store_true", help="Start webhook server") parser.add_argument("--port", type=int, default=5050, help="Server port") parser.add_argument("--test", action="store_true", help="Test with sample lead") args = parser.parse_args() if args.serve: print(f"🚀 Starting Lead Capture Webhook on port {args.port}") print(f" Lead endpoint: http://localhost:{args.port}/webhook/lead") print(f" Demo endpoint: http://localhost:{args.port}/webhook/demo-request") app.run(host="0.0.0.0", port=args.port, debug=False) elif args.test: # Test with sample lead sample_lead = { "name": "Test User", "email": "test@example.com", "phone": "+61400000000", "business_name": "Test Plumbing", "business_type": "trades", "missed_calls_week": 15, "avg_job_value": 200, "source": "test" } result = handler.process_lead(sample_lead) print(json.dumps(result, indent=2)) else: parser.print_help() if __name__ == "__main__": main()