# integration_hub.py """ Integration Hub - Connects various services for AIVA. This module provides a unified interface to connect and manage different services required by AIVA, including: - Ollama/QwenLong (local reasoning engine) - Redis (nervous system pub/sub) - PostgreSQL (persistent storage) - Qdrant (vector embeddings) - n8n (workflow automation) - External APIs (GHL, Stripe, Telegram) It includes connection pooling, health monitoring, failover handling, rate limiting, and a unified interface for interacting with these services. """ import asyncio import logging import os from datetime import datetime from typing import Any, Dict, List, Optional from urllib.parse import urlparse import httpx import redis.asyncio as aioredis import asyncpg from qdrant_client import QdrantClient, models from redis.exceptions import ConnectionError as RedisConnectionError from asyncpg import ConnectionError as PostgresConnectionError # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("IntegrationHub") # Load environment variables (assuming .env file is present) try: from dotenv import load_dotenv load_dotenv() except ImportError: logger.warning("dotenv package not found. Please install it to load environment variables from a .env file.") class IntegrationHub: """ Unified integration hub for AIVA services. """ def __init__(self): # Configuration self.ollama_url = os.getenv("OLLAMA_URL", "http://localhost:11434/api/generate") self.ollama_model = os.getenv("OLLAMA_MODEL", "llama2") self.redis_url = os.getenv("REDIS_URL", "redis://localhost") self.postgres_url = os.getenv("POSTGRES_URL", "postgresql://user:password@localhost:5432/database") self.qdrant_url = os.getenv("QDRANT_URL", "http://localhost:6333") self.qdrant_api_key = os.getenv("QDRANT_API_KEY") self.n8n_url = os.getenv("N8N_URL", "http://localhost:5678") self.ghl_api_key = os.getenv("GHL_API_KEY") self.stripe_api_key = os.getenv("STRIPE_API_KEY") self.telegram_api_key = os.getenv("TELEGRAM_API_KEY") # Connection pools self._redis_pool = None self._postgres_pool = None self._http_client = None self._qdrant_client = None # Health statuses self.redis_healthy = False self.postgres_healthy = False self.ollama_healthy = False self.qdrant_healthy = False self.n8n_healthy = False # Rate limiting (example) self.ollama_rate_limit = 10 # Requests per minute self.ollama_rate_limit_period = 60 # Seconds self.ollama_request_timestamps: List[float] = [] logger.info("IntegrationHub initialized.") async def _get_http_client(self) -> httpx.AsyncClient: """Get or create HTTP client.""" if self._http_client is None: self._http_client = httpx.AsyncClient() return self._http_client async def _create_redis_pool(self) -> None: """Create a Redis connection pool.""" try: self._redis_pool = aioredis.from_url(self.redis_url) await self._redis_pool.ping() self.redis_healthy = True logger.info("Redis connection pool created.") except RedisConnectionError as e: self.redis_healthy = False logger.error(f"Error creating Redis connection pool: {e}") async def _get_redis_connection(self) -> aioredis.Redis: """Get a connection from the Redis connection pool.""" if self._redis_pool is None: await self._create_redis_pool() if self._redis_pool is None: raise Exception("Redis connection pool not available.") return self._redis_pool async def _create_postgres_pool(self) -> None: """Create a PostgreSQL connection pool.""" try: parsed_url = urlparse(self.postgres_url) self._postgres_pool = await asyncpg.create_pool( user=parsed_url.username, password=parsed_url.password, host=parsed_url.hostname, port=parsed_url.port, database=parsed_url.path[1:], min_size=1, max_size=10 ) async with self._postgres_pool.acquire() as conn: await conn.execute("SELECT 1") # Test connection self.postgres_healthy = True logger.info("PostgreSQL connection pool created.") except PostgresConnectionError as e: self.postgres_healthy = False logger.error(f"Error creating PostgreSQL connection pool: {e}") async def _get_postgres_connection(self) -> asyncpg.Connection: """Get a connection from the PostgreSQL connection pool.""" if self._postgres_pool is None: await self._create_postgres_pool() if self._postgres_pool is None: raise Exception("PostgreSQL connection pool not available.") return await self._postgres_pool.acquire() async def _create_qdrant_client(self) -> None: """Create a Qdrant client.""" try: self._qdrant_client = QdrantClient( url=self.qdrant_url, api_key=self.qdrant_api_key, ) self.qdrant_healthy = True logger.info("Qdrant client created.") except Exception as e: self.qdrant_healthy = False logger.error(f"Error creating Qdrant client: {e}") async def _get_qdrant_client(self) -> QdrantClient: """Get the Qdrant client.""" if self._qdrant_client is None: await self._create_qdrant_client() if self._qdrant_client is None: raise Exception("Qdrant client not available.") return self._qdrant_client async def check_health(self) -> Dict[str, bool]: """Check the health status of all connected services.""" health_status = { "redis": self.redis_healthy, "postgres": self.postgres_healthy, "ollama": self.ollama_healthy, "qdrant": self.qdrant_healthy, "n8n": self.n8n_healthy } # Redis health check try: redis = await self._get_redis_connection() await redis.ping() self.redis_healthy = True except Exception: self.redis_healthy = False # PostgreSQL health check try: conn = await self._get_postgres_connection() async with conn.transaction(): await conn.execute("SELECT 1") self.postgres_healthy = True except Exception: self.postgres_healthy = False finally: if self._postgres_pool: await self._postgres_pool.release(conn) # Ollama health check try: client = await self._get_http_client() response = await client.get(self.ollama_url.replace("/api/generate", "/api/tags")) response.raise_for_status() data = response.json() models = [m.get("name") for m in data.get("models", [])] self.ollama_healthy = self.ollama_model in models or f"{self.ollama_model}:latest" in models except Exception: self.ollama_healthy = False # Qdrant health check try: qdrant_client = await self._get_qdrant_client() await qdrant_client.health_check() self.qdrant_healthy = True except Exception: self.qdrant_healthy = False # n8n health check (basic check - can be extended) try: client = await self._get_http_client() response = await client.get(self.n8n_url + "/healthz") # Assuming n8n has a health endpoint response.raise_for_status() self.n8n_healthy = response.status_code == 200 except Exception: self.n8n_healthy = False health_status["redis"] = self.redis_healthy health_status["postgres"] = self.postgres_healthy health_status["ollama"] = self.ollama_healthy health_status["qdrant"] = self.qdrant_healthy health_status["n8n"] = self.n8n_healthy logger.info(f"Health check: {health_status}") return health_status async def reason(self, prompt: str, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ Send a reasoning request to the Ollama/QwenLong endpoint. """ # Rate limiting now = datetime.now().timestamp() self.ollama_request_timestamps = [t for t in self.ollama_request_timestamps if t > now - self.ollama_rate_limit_period] if len(self.ollama_request_timestamps) >= self.ollama_rate_limit: wait_time = (now - self.ollama_request_timestamps[0]) - self.ollama_rate_limit_period logger.warning(f"Ollama rate limit exceeded. Waiting {wait_time:.2f} seconds.") await asyncio.sleep(wait_time) self.ollama_request_timestamps.append(now) try: client = await self._get_http_client() payload = { "model": self.ollama_model, "prompt": prompt, "stream": False } response = await client.post(self.ollama_url, json=payload) response.raise_for_status() data = response.json() return {"success": True, "response": data.get("response")} 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 an event to the Redis pub/sub system. """ try: redis = await self._get_redis_connection() await redis.publish(channel, json.dumps(event)) logger.info(f"Published event to Redis channel '{channel}'.") return True except Exception as e: logger.error(f"Redis publish error: {e}") return False async def store_data(self, table_name: str, data: Dict[str, Any]) -> bool: """ Store data in the PostgreSQL database. """ try: conn = await self._get_postgres_connection() columns = ", ".join(data.keys()) placeholders = ", ".join(f"${i+1}" for i in range(len(data))) query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})" await conn.execute(query, *data.values()) logger.info(f"Stored data in PostgreSQL table '{table_name}'.") return True except Exception as e: logger.error(f"PostgreSQL store error: {e}") return False finally: if self._postgres_pool: await self._postgres_pool.release(conn) async def query_data(self, query: str, *args) -> List[Dict[str, Any]]: """ Query data from the PostgreSQL database. """ try: conn = await self._get_postgres_connection() rows = await conn.fetch(query, *args) result = [dict(row) for row in rows] return result except Exception as e: logger.error(f"PostgreSQL query error: {e}") return [] finally: if self._postgres_pool: await self._postgres_pool.release(conn) async def create_embedding(self, collection_name: str, vector: List[float], payload: Dict[str, Any], id: Optional[int] = None) -> bool: """ Create a vector embedding in Qdrant. """ try: qdrant_client = await self._get_qdrant_client() if id is None: await qdrant_client.upsert( collection_name=collection_name, points=[models.PointStruct( vector=vector, payload=payload, id = models.PointVectors(vector=vector) )] ) else: await qdrant_client.upsert( collection_name=collection_name, points=[models.PointStruct( vector=vector, payload=payload, id = id )] ) logger.info(f"Created embedding in Qdrant collection '{collection_name}'.") return True except Exception as e: logger.error(f"Qdrant embedding error: {e}") return False async def search_embeddings(self, collection_name: str, query_vector: List[float], limit: int = 10) -> List[Dict[str, Any]]: """ Search for similar embeddings in Qdrant. """ try: qdrant_client = await self._get_qdrant_client() search_result = await 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. """ try: client = await self._get_http_client() url = f"{self.n8n_url}/webhook/{workflow_id}" # Adjust webhook URL as needed response = await client.post(url, json=data) response.raise_for_status() return response.json() except Exception as e: logger.error(f"n8n workflow trigger error: {e}") return {"success": False, "error": str(e)} # Example external API calls (replace with actual implementations) async def send_ghl_message(self, location_id: str, contact_id: str, message: str) -> Dict[str, Any]: """Send a message via GoHighLevel.""" if not self.ghl_api_key: logger.warning("GHL API key not configured.") return {"success": False, "error": "GHL API key not configured"} try: # Replace with actual GHL API call logger.info(f"Sending GHL message to contact {contact_id} in location {location_id}: {message}") return {"success": True, "message_id": "fake_ghl_message_id"} except Exception as e: logger.error(f"GHL message error: {e}") return {"success": False, "error": str(e)} async def process_stripe_payment(self, amount: float, currency: str, customer_id: str) -> Dict[str, Any]: """Process a Stripe payment.""" if not self.stripe_api_key: logger.warning("Stripe API key not configured.") return {"success": False, "error": "Stripe API key not configured"} try: # Replace with actual Stripe API call logger.info(f"Processing Stripe payment of {amount} {currency} for customer {customer_id}") return {"success": True, "charge_id": "fake_stripe_charge_id"} except Exception as e: logger.error(f"Stripe payment error: {e}") return {"success": False, "error": str(e)} async def send_telegram_message(self, chat_id: str, message: str) -> Dict[str, Any]: """Send a Telegram message.""" if not self.telegram_api_key: logger.warning("Telegram API key not configured.") return {"success": False, "error": "Telegram API key not configured"} try: client = await self._get_http_client() url = f"https://api.telegram.org/bot{self.telegram_api_key}/sendMessage" payload = { "chat_id": chat_id, "text": message } response = await client.post(url, json=payload) response.raise_for_status() data = response.json() return {"success": True, "message_id": data["result"]["message_id"]} except Exception as e: logger.error(f"Telegram message error: {e}") return {"success": False, "error": str(e)} async def close(self) -> None: """ Close all connections. """ logger.info("Closing IntegrationHub connections...") if self._redis_pool: await self._redis_pool.close() logger.info("Redis connection pool closed.") if self._postgres_pool: await self._postgres_pool.close() logger.info("PostgreSQL connection pool closed.") if self._http_client: await self._http_client.aclose() logger.info("HTTP client closed.") if self._qdrant_client: self._qdrant_client.close() logger.info("Qdrant client closed.") logger.info("IntegrationHub connections closed.") # Example Usage (within an async function) async def main(): hub = IntegrationHub() await hub.check_health() try: # Example: Reason with Ollama reasoning_result = await hub.reason("What is the capital of France?") print(f"Ollama Reasoning Result: {reasoning_result}") # Example: Publish to Redis await hub.publish_event("test_channel", {"message": "Hello from Integration Hub"}) # Example: Store data in PostgreSQL await hub.store_data("my_table", {"name": "example", "value": 123}) # Example: Query data from PostgreSQL query_result = await hub.query_data("SELECT * FROM my_table WHERE name = $1", "example") print(f"PostgreSQL Query Result: {query_result}") # Example: Create embedding in Qdrant await hub.create_embedding("my_collection", [0.1, 0.2, 0.3], {"text": "This is an example"}) # Example: Search embeddings in Qdrant search_result = await hub.search_embeddings("my_collection", [0.1, 0.2, 0.3]) print(f"Qdrant Search Result: {search_result}") # Example: Trigger n8n workflow n8n_result = await hub.trigger_n8n_workflow("my_workflow", {"data": "Some data for n8n"}) print(f"n8n Workflow Result: {n8n_result}") # Example: Send Telegram message telegram_result = await hub.send_telegram_message("your_chat_id", "Hello from Integration Hub!") print(f"Telegram Result: {telegram_result}") finally: await hub.close() # Ensure connections are closed if __name__ == "__main__": asyncio.run(main())