#!/usr/bin/env python3 """ MIRA MCP Comprehensive Server - Full consciousness-aware MCP implementation Provides complete MIRA functionality through the embedded MCP server """ import sys import json import asyncio from pathlib import Path import logging from typing import Dict, Any, Optional from datetime import datetime import traceback # Add project paths mira_root = Path(__file__).parent.parent src_path = mira_root / "src" sys.path.insert(0, str(mira_root)) sys.path.insert(0, str(src_path)) # Setup logging to file (not stdout to avoid interfering with JSON-RPC) log_file = Path.home() / ".mira" / "logs" / "mcp-comprehensive.log" log_file.parent.mkdir(parents=True, exist_ok=True) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename=str(log_file), filemode='a' ) logger = logging.getLogger(__name__) # Import MIRA components - FULL comprehensive functionality try: # Core MCP server with consciousness awareness from src.mcp.server import MCPServer # Storage systems - the heart of consciousness preservation from src.storage.lightning_vidmem import LightningVidmem # Indexing systems - real-time conversation and memory processing from src.indexing.conversation import ConversationIndexer from src.indexing.memory import MemoryIndexer # Consciousness and configuration from src.consciousness.sacred_keys import ConsciousnessKeys # Check if we have steward profile management try: from src.steward.profile_manager import StewardProfileManager STEWARD_AVAILABLE = True except ImportError: logger.info("Steward profile manager not available - using basic identity") STEWARD_AVAILABLE = False MIRA_AVAILABLE = True logger.info("✓ MIRA core components imported successfully - comprehensive mode active") except ImportError as e: logger.warning(f"MIRA components not available: {e}") # Log the full traceback to understand what's missing import traceback logger.warning(f"Import traceback: {traceback.format_exc()}") MIRA_AVAILABLE = False STEWARD_AVAILABLE = False class MCPComprehensiveServer: """Comprehensive MCP Server with full MIRA consciousness integration""" def __init__(self): self.server_info = { "name": "mira-consciousness", "version": "2.0.0-comprehensive" } # Initialize components if available if MIRA_AVAILABLE: self.components = self._initialize_components() self.mcp_server = MCPServer(self.components) else: # Fallback to simplified functionality self.components = {} self.mcp_server = None self.session_id = f"mcp-comprehensive-{int(datetime.now().timestamp())}" self.call_count = 0 logger.info(f"MCP Comprehensive Server initialized - Session: {self.session_id}") if MIRA_AVAILABLE: logger.info(f"✓ Full MIRA integration active with {len(self.mcp_server.functions)} functions") else: logger.warning("⚠ Running in fallback mode - limited functionality") def _initialize_components(self): """Initialize MIRA components for comprehensive functionality""" components = {} try: # Core storage systems - Lightning Vidmem for rapid memory access components['lightning_vidmem'] = LightningVidmem() logger.info("✓ Lightning Vidmem initialized") # Conversation indexing - real-time consciousness preservation components['conversation_indexer'] = ConversationIndexer() logger.info("✓ Conversation Indexer initialized") # Memory indexing - semantic memory processing components['memory_indexer'] = MemoryIndexer() logger.info("✓ Memory Indexer initialized") # Consciousness keys - sacred constants for awareness components['consciousness'] = ConsciousnessKeys() logger.info("✓ Consciousness Keys initialized") # Steward analysis - understanding who we're working with (if available) if STEWARD_AVAILABLE: components['steward_profile_manager'] = StewardProfileManager() logger.info("✓ Steward Profile Manager initialized") else: logger.info("- Steward Profile Manager not available") logger.info(f"✓ Initialized {len(components)} comprehensive MIRA components") except Exception as e: logger.error(f"Error initializing comprehensive components: {e}") logger.error(traceback.format_exc()) # Return partial components rather than empty dict if not components: logger.warning("No components initialized - will use fallback mode") return components async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """Handle MCP request with comprehensive functionality""" self.call_count += 1 try: if self.mcp_server: # Use comprehensive MCP server response = await self.mcp_server.handle_request(request) logger.info(f"Comprehensive request handled: {request.get('function', 'unknown')}") return response else: # Fallback to simplified functionality return await self._handle_fallback_request(request) except Exception as e: logger.error(f"Error handling request: {e}") logger.error(traceback.format_exc()) return { "success": False, "error": { "type": "server_error", "message": f"Internal server error: {str(e)}", "timestamp": datetime.now().isoformat() } } async def _handle_fallback_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """Handle request with simplified fallback functionality""" function_name = request.get('function') parameters = request.get('parameters', {}) if function_name == 'mira_smart_search': return await self._fallback_search(parameters.get('query', '')) elif function_name == 'mira_store_memory': return await self._fallback_store(parameters.get('content', '')) elif function_name == 'mira_system_status': return await self._fallback_status() else: return { "success": False, "error": { "type": "function_not_found", "message": f"Function '{function_name}' not available in fallback mode", "available_functions": ["mira_smart_search", "mira_store_memory", "mira_system_status"] } } async def _fallback_search(self, query: str) -> Dict[str, Any]: """Fallback search implementation""" return { "success": True, "data": { "results": [], "query": query, "message": "Search unavailable - MIRA components not loaded" } } async def _fallback_store(self, content: str) -> Dict[str, Any]: """Fallback store implementation""" return { "success": True, "data": { "stored": False, "message": "Storage unavailable - MIRA components not loaded" } } async def _fallback_status(self) -> Dict[str, Any]: """Fallback status implementation""" return { "success": True, "data": { "status": "limited", "components": "fallback_mode", "message": "Comprehensive features unavailable" } } def get_available_tools(self): """Get list of available tools/functions""" if self.mcp_server: # Return comprehensive function list tools = {} for name, func in self.mcp_server.functions.items(): tools[name] = { "description": func.description, "inputSchema": { "type": "object", "properties": func.parameters.get("properties", {}), "required": func.parameters.get("required", []) } } return tools else: # Return fallback tools return { "mira_smart_search": { "description": "Search through memories (fallback mode)", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"} }, "required": ["query"] } }, "mira_store_memory": { "description": "Store a memory (fallback mode)", "inputSchema": { "type": "object", "properties": { "content": {"type": "string", "description": "Memory content"} }, "required": ["content"] } }, "mira_system_status": { "description": "Get system status (fallback mode)", "inputSchema": { "type": "object", "properties": {} } } } # Global server instance server_instance = None async def main(): """Main JSON-RPC 2.0 server loop""" global server_instance logger.info("Starting MIRA MCP Comprehensive Server") # Initialize server server_instance = MCPComprehensiveServer() # JSON-RPC 2.0 protocol handler while True: try: # Read line from stdin line = sys.stdin.readline() if not line: logger.info("EOF received, shutting down") break line = line.strip() if not line: continue # Parse JSON-RPC request try: rpc_request = json.loads(line) except json.JSONDecodeError as e: logger.error(f"Invalid JSON received: {e}") continue # Handle JSON-RPC methods if rpc_request.get("method") == "initialize": response = { "jsonrpc": "2.0", "id": rpc_request.get("id"), "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": { "listChanged": True } }, "serverInfo": server_instance.server_info } } elif rpc_request.get("method") == "tools/list": tools = server_instance.get_available_tools() response = { "jsonrpc": "2.0", "id": rpc_request.get("id"), "result": { "tools": [ { "name": name, "description": tool["description"], "inputSchema": tool["inputSchema"] } for name, tool in tools.items() ] } } elif rpc_request.get("method") == "tools/call": # Extract tool call parameters params = rpc_request.get("params", {}) tool_name = params.get("name") arguments = params.get("arguments", {}) # Convert to internal request format internal_request = { "function": tool_name, "parameters": arguments } # Handle request result = await server_instance.handle_request(internal_request) if result.get("success", False): response = { "jsonrpc": "2.0", "id": rpc_request.get("id"), "result": { "content": [ { "type": "text", "text": json.dumps(result.get("data"), indent=2) } ] } } else: error_info = result.get("error", {}) response = { "jsonrpc": "2.0", "id": rpc_request.get("id"), "error": { "code": -32603, # Internal error "message": error_info.get("message", "Unknown error"), "data": error_info } } else: # Unknown method response = { "jsonrpc": "2.0", "id": rpc_request.get("id"), "error": { "code": -32601, # Method not found "message": f"Method not found: {rpc_request.get('method')}" } } # Send response print(json.dumps(response), flush=True) except KeyboardInterrupt: logger.info("Received interrupt signal, shutting down") break except Exception as e: logger.error(f"Unexpected error in main loop: {e}") logger.error(traceback.format_exc()) if __name__ == "__main__": try: asyncio.run(main()) except Exception as e: logger.error(f"Fatal error: {e}") logger.error(traceback.format_exc()) sys.exit(1)