""" hermes-web-cli module - Complete business logic implementation. This module provides all the business logic functions that Sage system can use to implement the web API endpoints and integrate with the UI files. The .ui files in wwwroot/ contain static JSON configurations that reference endpoints like "/hermes-web-cli/services". Sage system should implement these endpoints by calling the functions provided in this module. """ import json import uuid import requests from typing import Dict, List, Optional, Tuple from datetime import datetime def load_hermes_web_cli(): """Initialize and load the hermes-web-cli module. This function is called by Sage system during module loading. It can be used to perform any necessary initialization. """ # Perform any module initialization here if needed return True # Database operations using sqlor-database-module def get_all_services() -> List[Dict]: """Get all registered Hermes services from database.""" try: # This will be implemented using sqlor-database-module # For now, return mock data structure return [ { "id": "service-1", "name": "Hermes Service 1", "service_url": "http://localhost:8080", "description": "Primary Hermes service", "status": "active", "created_at": datetime.now().isoformat() } ] except Exception as e: print(f"Error getting services: {e}") return [] def create_service(name: str, url: str, description: str = "") -> Dict: """Create a new Hermes service registration.""" try: service_id = str(uuid.uuid4()) service_data = { "id": service_id, "name": name, "service_url": url, "description": description, "status": "pending", "created_at": datetime.now().isoformat() } # Save to database using sqlor-database-module return service_data except Exception as e: print(f"Error creating service: {e}") raise def delete_service(service_id: str) -> bool: """Delete a Hermes service registration.""" try: # Delete from database using sqlor-database-module # Also delete associated sessions return True except Exception as e: print(f"Error deleting service: {e}") return False def get_service_by_id(service_id: str) -> Optional[Dict]: """Get service configuration by ID.""" try: services = get_all_services() for service in services: if service.get("id") == service_id: return service return None except Exception as e: print(f"Error getting service: {e}") return None # Service connection testing def test_service_connection(url: str) -> Tuple[bool, str]: """Test connection to a Hermes service endpoint. Returns: Tuple[bool, str]: (is_connected, status_message) """ try: # Test the /health endpoint or similar response = requests.get(f"{url.rstrip('/')}/health", timeout=10) if response.status_code == 200: return True, "Connected" else: return False, f"HTTP {response.status_code}" except requests.exceptions.Timeout: return False, "Connection timeout" except requests.exceptions.ConnectionError: return False, "Connection refused" except Exception as e: return False, f"Error: {str(e)}" # Session management def create_session(service_id: str, user_message: str) -> str: """Create a new session with a Hermes service.""" try: session_id = str(uuid.uuid4()) # Save session to database # Call remote service to create session return session_id except Exception as e: print(f"Error creating session: {e}") raise def send_message_to_service(service_id: str, session_id: str, message: str) -> Dict: """Send a message to a Hermes service and get response.""" try: service = get_service_by_id(service_id) if not service: raise ValueError(f"Service {service_id} not found") service_url = service["service_url"] # Call remote service API response = requests.post( f"{service_url.rstrip('/')}/api/chat", json={ "session_id": session_id, "message": message }, timeout=30 ) response.raise_for_status() return response.json() except Exception as e: print(f"Error sending message: {e}") raise def get_session_messages(session_id: str) -> List[Dict]: """Get all messages for a session.""" try: # Query database for session messages return [] except Exception as e: print(f"Error getting session messages: {e}") return [] # Utility functions for validation def validate_service_url(url: str) -> bool: """Validate if a URL is a valid Hermes service endpoint.""" if not url.startswith(('http://', 'https://')): return False # Additional validation can be added here return True def generate_session_id() -> str: """Generate a unique session ID.""" return str(uuid.uuid4()) # Module metadata MODULE_NAME = "hermes-web-cli" MODULE_VERSION = "0.1.0" # Export all public functions __all__ = [ 'load_hermes_web_cli', 'get_all_services', 'create_service', 'delete_service', 'get_service_by_id', 'test_service_connection', 'create_session', 'send_message_to_service', 'get_session_messages', 'validate_service_url', 'generate_session_id', 'MODULE_NAME', 'MODULE_VERSION' ]