# integration_hub.py """ Integration Hub - Connects AIVA's Core Components ================================================= Connects Ollama/QwenLong, Redis, PostgreSQL, Qdrant, n8n, and external APIs (GHL, Stripe, Telegram). Provides: - Connection pooling - Health monitoring - Failover handling - Rate limiting - Unified interface """ import asyncio import json import logging import os from datetime import datetime from typing import Any, Dict, List, Optional import httpx import psycopg import redis.asyncio as aioredis from qdrant_client import QdrantClient, models from psycopg_pool import AsyncConnectionPool # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("IntegrationHub") # Load environment variables from dotenv import load_dotenv load_dotenv() class IntegrationHub: """ Unified hub for connecting and managing AIVA's core components. """ def __init__(self): # Configuration from environment variables self.ollama_url = os.getenv("OLLAMA_URL", "http://localhost:11434/api/generate") # Default Ollama URL self.redis_url = os.getenv("REDIS_URL", "redis://localhost:6379") # Default Redis URL self.postgres_url = os.getenv("POSTGRES_URL", "postgresql://user:password@localhost:5432/database") # Default PostgreSQL URL self.qdrant_url = os.getenv("QDRANT_URL", "http://localhost:6333") # Default Qdrant URL self.qdrant_api_key = os.getenv("QDRANT_API_KEY") self.n8n_url = os.getenv("N8N_URL", "http://localhost:5678/webhook/") # Default n8n URL self.ghl_api_key = os.getenv("GHL_API_KEY") # GoHighLevel API Key self.stripe_api_key = os.getenv("STRIPE_API_KEY") # Stripe API Key self.telegram_bot_token = os.getenv("TELEGRAM_BOT_TOKEN") # Telegram Bot Token # Connection pools and clients self._redis_client = None self._http_client = None self._postgres_pool = None self._qdrant_client = None # Health status self.health = { "ollama": "offline", "redis": "offline", "postgres": "offline", "qdrant": "offline", "n8n": "offline", "ghl": "offline", "stripe": "offline", "telegram": "offline" } # Rate limiting (example) self.api_call_counts = {} self.rate_limit_window = 60 # seconds self.max_api_calls = 100 self.model = os.getenv("AIVA_MODEL", "qwen-long") logger.info("IntegrationHub initialized.") async def connect(self): """Establish connections to all services.""" await asyncio.gather( self._connect_redis(), self._connect_postgres(), self._connect_qdrant(), self._test_n8n_connection(), self._test_ghl_connection(), self._test_stripe_connection(), self._test_telegram_connection(), self.ollama_health_check() ) logger.info("All services connected.") await self.update_health_status() async def close(self): """Close all connections.""" if self._redis_client: await self._redis_client.close() if self._http_client: await self._http_client.aclose() if self._postgres_pool: await self._postgres_pool.close() if self._qdrant_client: self._qdrant_client.close() # QdrantClient doesn't have async close logger.info("All connections closed.") # --- Connection Management --- async def _connect_redis(self): """Connect to Redis.""" try: self._redis_client = aioredis.from_url(self.redis_url) await self._redis_client.ping() self.health["redis"] = "online" logger.info("Connected to Redis.") except Exception as e: self.health["redis"] = "offline" logger.error(f"Redis connection error: {e}") async def _connect_postgres(self): """Connect to PostgreSQL using connection pooling.""" try: self._postgres_pool = AsyncConnectionPool(self.postgres_url) async with self._postgres_pool.connection() as conn: async with conn.cursor() as cur: await cur.execute("SELECT 1;") self.health["postgres"] = "online" logger.info("Connected to PostgreSQL.") except Exception as e: self.health["postgres"] = "offline" logger.error(f"PostgreSQL connection error: {e}") async def _connect_qdrant(self): """Connect to Qdrant.""" try: self._qdrant_client = QdrantClient(url=self.qdrant_url, api_key=self.qdrant_api_key) self._qdrant_client.get_telemetry_data() # Test connection self.health["qdrant"] = "online" logger.info("Connected to Qdrant.") except Exception as e: self.health["qdrant"] = "offline" logger.error(f"Qdrant connection error: {e}") async def _test_n8n_connection(self): """Test connection to n8n by sending a simple request.""" try: async with httpx.AsyncClient() as client: response = await client.post(self.n8n_url + "health_check", json={"status": "checking"}, timeout=10) response.raise_for_status() self.health["n8n"] = "online" logger.info("Connected to n8n.") except Exception as e: self.health["n8n"] = "offline" logger.error(f"n8n connection error: {e}") async def _test_ghl_connection(self): """Test connection to GoHighLevel API.""" if not self.ghl_api_key: self.health["ghl"] = "unconfigured" logger.warning("GoHighLevel API key not configured.") return try: async with httpx.AsyncClient() as client: headers = {"Authorization": f"Bearer {self.ghl_api_key}"} response = await client.get("https://services.leadconnectorhq.com/service/v1/users/me", headers=headers, timeout=10) response.raise_for_status() self.health["ghl"] = "online" logger.info("Connected to GoHighLevel.") except Exception as e: self.health["ghl"] = "offline" logger.error(f"GoHighLevel connection error: {e}") async def _test_stripe_connection(self): """Test connection to Stripe API.""" if not self.stripe_api_key: self.health["stripe"] = "unconfigured" logger.warning("Stripe API key not configured.") return try: async with httpx.AsyncClient() as client: headers = {"Authorization": f"Bearer {self.stripe_api_key}"} response = await client.get("https://api.stripe.com/v1/charges", headers=headers, timeout=10) response.raise_for_status() self.health["stripe"] = "online" logger.info("Connected to Stripe.") except Exception as e: self.health["stripe"] = "offline" logger.error(f"Stripe connection error: {e}") async def _test_telegram_connection(self): """Test connection to Telegram Bot API.""" if not self.telegram_bot_token: self.health["telegram"] = "unconfigured" logger.warning("Telegram Bot Token not configured.") return try: async with httpx.AsyncClient() as client: url = f"https://api.telegram.org/bot{self.telegram_bot_token}/getMe" response = await client.get(url, timeout=10) response.raise_for_status() self.health["telegram"] = "online" logger.info("Connected to Telegram.") except Exception as e: self.health["telegram"] = "offline" logger.error(f"Telegram connection error: {e}") async def _get_http_client(self) -> httpx.AsyncClient: """Get or create HTTP client.""" if self._http_client is None or self._http_client.is_closed: self._http_client = httpx.AsyncClient() return self._http_client # --- Health Monitoring --- async def ollama_health_check(self) -> Dict[str, Any]: """Check AIVA/Ollama health status.""" tags_url = self.ollama_url.replace("/api/generate", "/api/tags") try: async with httpx.AsyncClient() as client: response = await client.get(tags_url, timeout=10.0) response.raise_for_status() data = response.json() models_available = [m.get("name") for m in data.get("models", [])] qwen_ready = self.model in models_available or f"{self.model}:latest" in models_available if qwen_ready: self.health["ollama"] = "online" else: self.health["ollama"] = "model_missing" return { "status": self.health["ollama"], "models": models_available, "target_model": self.model, "endpoint": self.ollama_url.split("/api")[0], "timestamp": datetime.now().isoformat() } except Exception as e: self.health["ollama"] = "offline" return { "status": "offline", "error": str(e), "endpoint": self.ollama_url.split("/api")[0], "timestamp": datetime.now().isoformat() } async def update_health_status(self): """Aggregate and log health status.""" logger.info(f"Health Status: {self.health}") return self.health # --- Unified Interface Methods --- async def reason(self, prompt: str, context: Optional[Dict[str, Any]] = None, **kwargs) -> Dict[str, Any]: """ Send reasoning request to AIVA (Ollama). """ if self.health["ollama"] != "online": return {"success": False, "error": "Ollama is offline."} payload = { "model": self.model, "prompt": prompt, "stream": False, } try: async with httpx.AsyncClient() as client: response = await client.post(self.ollama_url, json=payload) response.raise_for_status() data = response.json() return { "success": True, "response": data.get("response", ""), "model": self.model } except Exception as e: logger.error(f"Ollama reasoning error: {e}") return {"success": False, "error": str(e)} async def publish_event(self, channel: str, event: Dict[str, Any]) -> bool: """Publish event to Redis.""" if self.health["redis"] != "online": logger.warning("Redis is offline, cannot publish event.") return False try: event["timestamp"] = datetime.now().isoformat() event["source"] = "integration_hub" await self._redis_client.publish(channel, json.dumps(event)) logger.info(f"Published to {channel}: {event.get('type', 'unknown')}") return True except Exception as e: logger.error(f"Redis publish error: {e}") return False async def execute_query(self, query: str, params: tuple = None) -> List[Dict[str, Any]]: """Execute a SQL query against PostgreSQL.""" if self.health["postgres"] != "online": logger.warning("PostgreSQL is offline, cannot execute query.") return [] try: async with self._postgres_pool.connection() as conn: async with conn.cursor() as cur: await cur.execute(query, params) if cur.description: columns = [col.name for col in cur.description] results = [dict(zip(columns, row)) for row in await cur.fetchall()] else: results = [] return results except Exception as e: logger.error(f"PostgreSQL query error: {e}") return [] async def store_embedding(self, collection_name: str, vector: List[float], metadata: Dict[str, Any], id: str = None): """Store a vector embedding in Qdrant.""" if self.health["qdrant"] != "online": logger.warning("Qdrant is offline, cannot store embedding.") return False try: if id: points = [models.PointStruct(id=id, vector=vector, payload=metadata)] else: points = [models.PointStruct(id=str(hash(json.dumps(metadata))), vector=vector, payload=metadata)] self._qdrant_client.upsert(collection_name=collection_name, points=points, wait=True) return True except Exception as e: logger.error(f"Qdrant store error: {e}") return False async def search_embedding(self, collection_name: str, query_vector: List[float], limit: int = 10) -> List[Dict[str, Any]]: """Search Qdrant for similar embeddings.""" if self.health["qdrant"] != "online": logger.warning("Qdrant is offline, cannot search embeddings.") return [] try: search_result = self._qdrant_client.search( collection_name=collection_name, query_vector=query_vector, limit=limit ) results = [hit.payload for hit in search_result] return results except Exception as e: logger.error(f"Qdrant search error: {e}") return [] async def trigger_n8n_workflow(self, workflow_id: str, data: Dict[str, Any]) -> Dict[str, Any]: """Trigger an n8n workflow with the given data.""" if self.health["n8n"] != "online": logger.warning("n8n is offline, cannot trigger workflow.") return {"status": "offline", "error": "n8n is offline"} try: async with httpx.AsyncClient() as client: response = await client.post(self.n8n_url + workflow_id, json=data, timeout=30) response.raise_for_status() return response.json() except Exception as e: logger.error(f"n8n workflow error: {e}") return {"status": "error", "error": str(e)} async def ghl_api_request(self, endpoint: str, method: str = "GET", data: Dict[str, Any] = None) -> Dict[str, Any]: """Make a request to the GoHighLevel API.""" if self.health["ghl"] != "online": return {"status": "offline", "error": "GoHighLevel API is offline or unconfigured."} try: async with httpx.AsyncClient() as client: headers = {"Authorization": f"Bearer {self.ghl_api_key}"} url = f"https://services.leadconnectorhq.com/service/v1/{endpoint}" if method == "GET": response = await client.get(url, headers=headers, timeout=30) elif method == "POST": response = await client.post(url, headers=headers, json=data, timeout=30) elif method == "PUT": response = await client.put(url, headers=headers, json=data, timeout=30) elif method == "DELETE": response = await client.delete(url, headers=headers, timeout=30) else: return {"status": "error", "error": "Invalid HTTP method."} response.raise_for_status() return response.json() except Exception as e: logger.error(f"GoHighLevel API error: {e}") return {"status": "error", "error": str(e)} async def stripe_api_request(self, endpoint: str, method: str = "GET", data: Dict[str, Any] = None) -> Dict[str, Any]: """Make a request to the Stripe API.""" if self.health["stripe"] != "online": return {"status": "offline", "error": "Stripe API is offline or unconfigured."} try: async with httpx.AsyncClient() as client: headers = {"Authorization": f"Bearer {self.stripe_api_key}"} url = f"https://api.stripe.com/v1/{endpoint}" if method == "GET": response = await client.get(url, headers=headers, timeout=30) elif method == "POST": response = await client.post(url, headers=headers, json=data, timeout=30) elif method == "PUT": response = await client.put(url, headers=headers, json=data, timeout=30) elif method == "DELETE": response = await client.delete(url, headers=headers, timeout=30) else: return {"status": "error", "error": "Invalid HTTP method."} response.raise_for_status() return response.json() except Exception as e: logger.error(f"Stripe API error: {e}") return {"status": "error", "error": str(e)} async def send_telegram_message(self, chat_id: str, text: str) -> Dict[str, Any]: """Send a message to a Telegram chat.""" if self.health["telegram"] != "online": return {"status": "offline", "error": "Telegram API is offline or unconfigured."} try: async with httpx.AsyncClient() as client: url = f"https://api.telegram.org/bot{self.telegram_bot_token}/sendMessage" data = {"chat_id": chat_id, "text": text} response = await client.post(url, json=data, timeout=30) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Telegram API error: {e}") return {"status": "error", "error": str(e)} # --- Rate Limiting --- def is_rate_limited(self, api_name: str) -> bool: """Check if the API call is rate limited.""" now = datetime.now().timestamp() if api_name not in self.api_call_counts: self.api_call_counts[api_name] = [] # Remove old timestamps self.api_call_counts[api_name] = [ts for ts in self.api_call_counts[api_name] if now - ts < self.rate_limit_window] if len(self.api_call_counts[api_name]) >= self.max_api_calls: return True self.api_call_counts[api_name].append(now) return False # --- Failover Handling --- # Implement failover logic here, e.g., using backup servers or local caches. # Example: # if self.health["redis"] == "offline": # # Use a local cache or backup Redis instance # Example Usage (Conceptual) async def main(): hub = IntegrationHub() await hub.connect() # Example 1: Reasoning with Ollama response = await hub.reason("What is the capital of France?") print(f"Ollama Response: {response}") # Example 2: Publishing an event to Redis event_data = {"type": "user_login", "user_id": "123"} await hub.publish_event("user_events", event_data) # Example 3: Executing a PostgreSQL query query = "SELECT * FROM users WHERE id = %s;" results = await hub.execute_query(query, ("123",)) print(f"PostgreSQL Results: {results}") # Example 4: Storing and searching embeddings in Qdrant embedding_vector = [0.1, 0.2, 0.3] metadata = {"text": "Example document"} await hub.store_embedding("my_collection", embedding_vector, metadata) search_results = await hub.search_embedding("my_collection", embedding_vector) print(f"Qdrant Search Results: {search_results}") # Example 5: Triggering an n8n workflow n8n_data = {"message": "Hello from AIVA!"} n8n_response = await hub.trigger_n8n_workflow("my_workflow", n8n_data) print(f"n8n Response: {n8n_response}") # Example 6: Making a GoHighLevel API request ghl_response = await hub.ghl_api_request("contacts") print(f"GoHighLevel Response: {ghl_response}") # Example 7: Making a Stripe API request stripe_response = await hub.stripe_api_request("charges") print(f"Stripe Response: {stripe_response}") # Example 8: Sending a Telegram message telegram_response = await hub.send_telegram_message("123456789", "Hello from AIVA!") print(f"Telegram Response: {telegram_response}") await hub.close() if __name__ == "__main__": asyncio.run(main())