#!/usr/bin/env python3 """ AIVA Real-Time Communication Bridge ==================================== FastAPI server for instant communication with Kinan and leads. Handles: Slack, VAPI voice, SMS (Telnyx), Web chat """ import os import json import asyncio import logging from datetime import datetime from pathlib import Path from typing import Optional import urllib.request import urllib.error # FastAPI try: from fastapi import FastAPI, Request, HTTPException from fastapi.responses import JSONResponse import uvicorn HAS_FASTAPI = True except ImportError: HAS_FASTAPI = False print("Install: pip install fastapi uvicorn") # Redis try: import redis HAS_REDIS = True except ImportError: HAS_REDIS = False # Load secrets SECRETS_PATH = Path("/mnt/e/genesis-system/config/secrets.env") if SECRETS_PATH.exists(): with open(SECRETS_PATH) as f: for line in f: if "=" in line and not line.startswith("#"): k, v = line.strip().split("=", 1) os.environ[k] = v.strip('"') # Config REDIS_CONFIG = { "host": os.environ.get("GENESIS_REDIS_HOST", "redis-genesis-u50607.vm.elestio.app"), "port": int(os.environ.get("GENESIS_REDIS_PORT", 26379)), "password": os.environ.get("GENESIS_REDIS_PASSWORD", ""), "decode_responses": True, } OLLAMA_URL = "http://152.53.201.152:23405/api/generate" AIVA_MODEL = "qwen-long" logging.basicConfig(level=logging.INFO, format='%(asctime)s - AIVA Bridge - %(message)s') logger = logging.getLogger("AIVABridge") app = FastAPI(title="AIVA Real-Time Bridge", version="1.0") class AIVABrain: """Quick access to AIVA's reasoning.""" @staticmethod async def think(prompt: str, max_tokens: int = 500) -> str: """Get AIVA's response.""" payload = { "model": AIVA_MODEL, "prompt": prompt, "stream": False, "options": {"num_ctx": 8192, "temperature": 0.7} } try: req = urllib.request.Request( OLLAMA_URL, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}, method="POST" ) with urllib.request.urlopen(req, timeout=60) as resp: data = json.loads(resp.read().decode()) return data.get("response", "I'm thinking...") except Exception as e: logger.error(f"AIVA think error: {e}") return "Sorry, I'm having trouble thinking right now. Let me get back to you." @app.get("/") async def root(): return {"status": "online", "service": "AIVA Real-Time Bridge", "timestamp": datetime.now().isoformat()} @app.get("/health") async def health(): """Health check endpoint.""" return { "status": "healthy", "redis": HAS_REDIS, "uptime": datetime.now().isoformat() } @app.post("/webhook/slack") async def slack_webhook(request: Request): """ Handle incoming Slack messages. Slack sends: challenge for verification, or event for messages. """ data = await request.json() # Slack URL verification challenge if "challenge" in data: return {"challenge": data["challenge"]} # Handle message events event = data.get("event", {}) if event.get("type") == "message" and not event.get("bot_id"): user = event.get("user", "unknown") text = event.get("text", "") channel = event.get("channel", "") logger.info(f"Slack message from {user}: {text[:50]}") # Get AIVA's response prompt = f"""You are AIVA, Kinan's AI assistant. Respond helpfully and concisely. User message: {text} Respond naturally as AIVA:""" response = await AIVABrain.think(prompt) # Store for async reply (n8n will pick up and send to Slack) if HAS_REDIS: r = redis.Redis(**REDIS_CONFIG) r.lpush("genesis:aiva_slack_replies", json.dumps({ "channel": channel, "text": response, "thread_ts": event.get("ts"), "timestamp": datetime.now().isoformat() })) return {"ok": True, "response": response} return {"ok": True} @app.post("/webhook/vapi") async def vapi_webhook(request: Request): """ Handle VAPI voice assistant webhooks. Called when a voice call needs AIVA's input. """ data = await request.json() message_type = data.get("type", "") if message_type == "function-call": # VAPI is asking AIVA to process something function_name = data.get("functionCall", {}).get("name", "") parameters = data.get("functionCall", {}).get("parameters", {}) logger.info(f"VAPI function call: {function_name}") if function_name == "askAIVA": question = parameters.get("question", "") response = await AIVABrain.think(f"A caller is asking: {question}\n\nRespond as AIVA:") return {"result": response} elif message_type == "transcript": # Live transcript from call transcript = data.get("transcript", "") logger.info(f"VAPI transcript: {transcript[:100]}") return {"ok": True} @app.post("/webhook/telnyx") async def telnyx_webhook(request: Request): """ Handle Telnyx SMS/voice webhooks. """ data = await request.json() event_type = data.get("data", {}).get("event_type", "") if event_type == "message.received": # Incoming SMS payload = data.get("data", {}).get("payload", {}) from_number = payload.get("from", {}).get("phone_number", "") text = payload.get("text", "") logger.info(f"SMS from {from_number}: {text[:50]}") # Get AIVA's response response = await AIVABrain.think(f"SMS from lead: {text}\n\nRespond as AIVA (keep under 160 chars):") # Queue reply if HAS_REDIS: r = redis.Redis(**REDIS_CONFIG) r.lpush("genesis:aiva_sms_replies", json.dumps({ "to": from_number, "text": response[:160], "timestamp": datetime.now().isoformat() })) return {"ok": True, "response": response} return {"ok": True} @app.post("/chat") async def direct_chat(request: Request): """ Direct chat endpoint for web interface or API calls. """ data = await request.json() message = data.get("message", "") user = data.get("user", "Kinan") if not message: raise HTTPException(status_code=400, detail="Message required") logger.info(f"Chat from {user}: {message[:50]}") prompt = f"""You are AIVA, the autonomous AI assistant for Genesis-OS. You work for Kinan and help manage leads, tasks, and business operations. {user} says: {message} Respond as AIVA:""" response = await AIVABrain.think(prompt) return { "from": "AIVA", "message": response, "timestamp": datetime.now().isoformat() } @app.post("/message/kinan") async def message_kinan(request: Request): """ AIVA proactively messages Kinan. Queues message for Slack delivery. """ data = await request.json() message = data.get("message", "") priority = data.get("priority", "normal") if HAS_REDIS: r = redis.Redis(**REDIS_CONFIG) r.lpush("genesis:aiva_to_kinan", json.dumps({ "message": message, "priority": priority, "timestamp": datetime.now().isoformat() })) logger.info(f"Queued message to Kinan: {message[:50]}") return {"ok": True, "queued": True} def main(): if not HAS_FASTAPI: print("ERROR: FastAPI not installed. Run: pip install fastapi uvicorn") return print(""" ╔═══════════════════════════════════════════════════════════╗ ║ AIVA REAL-TIME BRIDGE ║ ║ Slack | VAPI | Telnyx | Web Chat ║ ╚═══════════════════════════════════════════════════════════╝ Endpoints: - POST /webhook/slack - Slack events - POST /webhook/vapi - VAPI voice - POST /webhook/telnyx - Telnyx SMS - POST /chat - Direct chat API - GET /health - Health check """) uvicorn.run(app, host="0.0.0.0", port=8765, log_level="info") if __name__ == "__main__": main()