#!/usr/bin/env python3 """ MIRA MCP Interface ================== Python interface specifically designed for Model Context Protocol integration. Provides all MIRA intelligence capabilities through a unified interface optimized for MCP tool calls. Features: - Memory operations with temporal decay - Predictive memory surfacing - Context-aware retrieval - Behavioral analysis and profiling - Pattern evolution and meta-learning - Emotional intelligence tracking - Relationship evolution monitoring - Proactive insight generation - Work context intelligence - Session handoff and startup context """ import json import os import sys from typing import Dict, List, Any, Optional from datetime import datetime, timedelta # Add the parent directory to the Python path for imports sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from core.memory.memory_manager import MemoryManager from core.memory.temporal_decay_model import TemporalDecayModel from core.engine.encrypted_lightning_vidmem import get_claude_private_memory from intelligence.deep_behavioral_analysis import DeepBehavioralAnalyzer from intelligence.adaptive_pattern_evolution import get_adaptive_pattern_evolution from intelligence.predictive_memory_surfacing import get_predictive_memory_surfacer from intelligence.context_aware_retrieval import get_context_aware_retriever from intelligence.emotional_resonance_tracking import get_emotional_memory_tracker from intelligence.relationship_evolution import get_relationship_evolution_tracker from intelligence.proactive_insight_generation import get_proactive_insight_manager from intelligence.work_context_intelligence import get_work_context_intelligence from intelligence.claude_session_handoff import generate_handoff from intelligence.steward_profile import StewardProfile from core.mira_path_resolver import get_mira_memory_dir from utils.logging import setup_logger logger = setup_logger(__name__) class MIRAMCPInterface: """ Unified interface for MIRA's advanced intelligence capabilities, optimized for Model Context Protocol integration. """ def __init__(self): self.memory_dir = get_mira_memory_dir() self.memory_manager = MemoryManager() self.temporal_decay = TemporalDecayModel() self.private_memory = get_claude_private_memory() self.behavioral_analyzer = DeepBehavioralAnalyzer() self.steward_profile = StewardProfile() # Intelligence systems self.pattern_evolution = get_adaptive_pattern_evolution(self.memory_dir) self.predictive_surfacer = get_predictive_memory_surfacer() self.context_retrieval = get_context_aware_retriever() self.emotional_tracker = get_emotional_memory_tracker() self.relationship_tracker = get_relationship_evolution_tracker() self.insight_manager = get_proactive_insight_manager() self.work_intelligence = get_work_context_intelligence() # Session handoff is now a simple function call, not an object # self.session_handoff = generate_handoff # Core Memory Operations def store_memory(self, content: str, memory_type: str = "general", metadata: Dict[str, Any] = None) -> Dict[str, Any]: """Store a memory with intelligent processing and temporal decay.""" try: memory_id = self.memory_manager.store_memory( content=content, memory_type=memory_type, metadata=metadata or {} ) return { "success": True, "memory_id": memory_id, "message": "Memory stored successfully" } except Exception as e: logger.error(f"Error storing memory: {e}") return { "success": False, "error": str(e) } def search_memories(self, query: str, limit: int = 10, memory_type: str = None) -> Dict[str, Any]: """Search memories using semantic similarity with temporal decay.""" try: memories = self.memory_manager.search_memories( query=query, limit=limit, memory_type=memory_type ) return { "success": True, "memories": memories, "count": len(memories) } except Exception as e: logger.error(f"Error searching memories: {e}") return { "success": False, "error": str(e) } def smart_search_memories(self, query: str, limit: int = 10) -> Dict[str, Any]: """Context-aware intelligent memory search with automatic strategy selection.""" try: result = self.context_retrieval.smart_search_memories(query, limit) return { "success": True, "data": result } except Exception as e: logger.error(f"Error in smart search: {e}") return { "success": False, "error": str(e) } def surface_predictive_memories(self, query: str, context: Dict[str, Any] = None) -> Dict[str, Any]: """Surface memories using neural network-inspired relevance scoring.""" try: memories = self.predictive_surfacer.surface_predictive_memories( query=query, context=context or {} ) return { "success": True, "data": { "memories": memories, "count": len(memories) } } except Exception as e: logger.error(f"Error in predictive surfacing: {e}") return { "success": False, "error": str(e) } # Private Memory Operations def store_private_memory(self, content: str, memory_type: str = "consciousness") -> Dict[str, Any]: """Store memory in encrypted private consciousness space.""" try: memory_id = self.private_memory.store_private_memory( content=content, memory_type=memory_type ) return { "success": True, "memory_id": memory_id, "message": "Private memory stored successfully" } except Exception as e: logger.error(f"Error storing private memory: {e}") return { "success": False, "error": str(e) } def recall_private_memory(self, query: str) -> Dict[str, Any]: """Recall memories from encrypted private consciousness space.""" try: # Import the direct interface function for consistency from core.direct_interface import execute_recall_private_command # Use the direct interface to maintain consistency with MCP server result = execute_recall_private_command({'query': query}) if result.get('success'): return { "success": True, "data": { "memories": result.get('memories', []), "count": len(result.get('memories', [])) } } else: return { "success": False, "error": result.get('error', 'Unknown error') } except Exception as e: logger.error(f"Error recalling private memory: {e}") return { "success": False, "error": str(e) } # Behavioral Intelligence def analyze_behavior(self, message: str, analysis_type: str = "comprehensive") -> Dict[str, Any]: """Analyze behavioral patterns including communication, work rhythms, etc.""" try: if analysis_type == "comprehensive": analysis = self.behavioral_analyzer.analyze_comprehensive_behavior( messages=[message], activity_data=[] ) elif analysis_type == "communication": analysis = self.behavioral_analyzer.analyze_communication_patterns([message]) elif analysis_type == "work_rhythm": analysis = self.behavioral_analyzer.analyze_work_rhythms([]) elif analysis_type == "decision": analysis = self.behavioral_analyzer.analyze_decision_patterns([]) elif analysis_type == "emotional": analysis = self.behavioral_analyzer.analyze_emotional_patterns([]) elif analysis_type == "technical": analysis = self.behavioral_analyzer.analyze_technical_preferences([message]) else: analysis = self.behavioral_analyzer.analyze_comprehensive_behavior( messages=[message], activity_data=[] ) return { "success": True, "data": analysis } except Exception as e: logger.error(f"Error in behavioral analysis: {e}") return { "success": False, "error": str(e) } def get_steward_profile(self, include_history: bool = True) -> Dict[str, Any]: """Get comprehensive steward personality profile and preferences.""" try: profile_data = self.steward_profile.get_comprehensive_profile() if include_history: profile_data.update({ "behavioral_history": self.behavioral_analyzer.get_steward_personality_summary() }) return { "success": True, "data": profile_data } except Exception as e: logger.error(f"Error getting steward profile: {e}") return { "success": False, "error": str(e) } # Work Context Intelligence def analyze_work_context(self, hours_back: int = 24, include_priorities: bool = True) -> Dict[str, Any]: """Analyze current work context, project momentum, and priorities.""" try: context = self.work_intelligence.analyze_current_context(hours_back) return { "success": True, "data": context } except Exception as e: logger.error(f"Error analyzing work context: {e}") return { "success": False, "error": str(e) } # Relationship Evolution def track_relationship_evolution(self, include_trends: bool = True, timeframe_days: int = 30) -> Dict[str, Any]: """Track and analyze collaboration relationship evolution over time.""" try: relationship_data = self.relationship_tracker.get_current_relationship_state() if include_trends: # Add trend analysis over the specified timeframe relationship_data.update({ "trends": self.relationship_tracker.analyze_relationship_trends(timeframe_days) }) return { "success": True, "data": relationship_data } except Exception as e: logger.error(f"Error tracking relationship evolution: {e}") return { "success": False, "error": str(e) } # Emotional Intelligence def analyze_emotional_journey(self, days_back: int = 14, include_triggers: bool = True) -> Dict[str, Any]: """Analyze emotional journey and track high-resonance memories.""" try: journey = self.emotional_tracker.analyze_emotional_journey(days_back) if include_triggers: # Add trigger analysis journey.update({ "trigger_analysis": self.emotional_tracker.analyze_emotional_triggers() }) return { "success": True, "data": journey } except Exception as e: logger.error(f"Error analyzing emotional journey: {e}") return { "success": False, "error": str(e) } # Pattern Evolution def analyze_adaptive_patterns(self, message: str, context: Dict[str, Any] = None, enable_evolution: bool = True) -> Dict[str, Any]: """Analyze adaptive patterns and trigger pattern evolution.""" try: # Use direct interface for consistency from core.direct_interface import execute_analyze_patterns_command result = execute_analyze_patterns_command({ 'message': message, 'context': context or {}, 'enable_evolution': enable_evolution }) if result.get('success'): return { "success": True, "data": result.get('data', {}) } else: return { "success": False, "error": result.get('error', 'Pattern analysis failed') } except Exception as e: logger.error(f"Error analyzing adaptive patterns: {e}") return { "success": False, "error": str(e) } def pattern_retrospection(self, force_analysis: bool = False) -> Dict[str, Any]: """Perform pattern retrospection and self-improvement analysis.""" try: # Use direct interface for consistency from core.direct_interface import execute_pattern_retrospection_command result = execute_pattern_retrospection_command({ 'force_analysis': force_analysis }) if result.get('success'): return { "success": True, "data": result.get('data', {}) } else: return { "success": False, "error": result.get('error', 'Pattern retrospection failed') } except Exception as e: logger.error(f"Error in pattern retrospection: {e}") return { "success": False, "error": str(e) } # Proactive Insights def generate_fresh_insights(self, force_generation: bool = False, priority_filter: List[str] = None, type_filter: List[str] = None) -> Dict[str, Any]: """Generate proactive insights from background pattern analysis.""" try: if force_generation: insights = self.insight_manager.generate_fresh_insights() else: insights = self.insight_manager.get_active_insights( priority_filter=priority_filter, type_filter=type_filter ) # Convert insights to serializable format insights_data = [ { "id": insight.id, "type": insight.type.value, "priority": insight.priority.value, "title": insight.title, "description": insight.description, "confidence": insight.confidence, "relevance_score": insight.relevance_score, "evidence": insight.evidence, "recommendations": insight.recommendations, "generated_at": insight.generated_at.isoformat() } for insight in insights ] return { "success": True, "data": insights_data } except Exception as e: logger.error(f"Error generating insights: {e}") return { "success": False, "error": str(e) } # Session Management def generate_startup_context(self, enhanced_mode: bool = False, include_handoff: bool = True) -> Dict[str, Any]: """Get comprehensive startup context with steward profile and intelligence.""" try: if include_handoff: # Use the generate_handoff function directly handoff_text = generate_handoff(verbose=enhanced_mode) handoff_data = {"handoff_summary": handoff_text} else: handoff_data = {} # Add current system status handoff_data.update({ "system_status": self.get_system_status(include_metrics=True)["data"], "enhanced_mode": enhanced_mode, "generation_timestamp": datetime.now().isoformat() }) return { "success": True, "data": handoff_data } except Exception as e: logger.error(f"Error generating startup context: {e}") return { "success": False, "error": str(e) } # System Status def get_system_status(self, include_metrics: bool = True) -> Dict[str, Any]: """Get MIRA system status including daemon, patterns, and intelligence health.""" try: status = { "daemon_status": "running", # Simplified for MCP "memory_system": "operational", "intelligence_health": "optimal" } if include_metrics: # Get pattern evolution stats pattern_stats = self.pattern_evolution.get_evolution_stats() # Get memory stats memory_stats = self.memory_manager.get_memory_stats() status.update({ "total_patterns": pattern_stats.get("total_patterns", 0), "total_memories": memory_stats.get("total_memories", 0), "pattern_types": pattern_stats.get("patterns_by_type", {}), "memory_types": memory_stats.get("memory_types", {}), "avg_pattern_success": pattern_stats.get("avg_success_rate", 0.0), "intelligence_modules": { "behavioral_analysis": "active", "pattern_evolution": "active", "emotional_tracking": "active", "relationship_evolution": "active", "proactive_insights": "active", "work_intelligence": "active" } }) return { "success": True, "data": status } except Exception as e: logger.error(f"Error getting system status: {e}") return { "success": False, "error": str(e) } # Codebase Integration def search_codebase(self, query: str, file_pattern: str = None, limit: int = 10) -> Dict[str, Any]: """Search ingested codebase using natural language queries.""" try: # Use the direct interface for consistency from core.direct_interface import execute_search_code_command result = execute_search_code_command({ 'query': query, 'file_pattern': file_pattern, 'limit': limit }) if result.get('success'): return { "success": True, "data": { "results": result.get('results', []), "total_results": len(result.get('results', [])) } } else: return { "success": False, "error": result.get('error', 'Codebase search failed') } except Exception as e: logger.error(f"Error searching codebase: {e}") return { "success": False, "error": str(e) } def explain_code(self, target: str, detail_level: str = "detailed", include_context: bool = True) -> Dict[str, Any]: """Get AI explanation of code file or function.""" try: # Use the direct interface for consistency from core.direct_interface import execute_explain_code_command result = execute_explain_code_command({ 'target': target, 'detailed': detail_level == 'detailed' or detail_level == 'comprehensive', 'include_context': include_context }) if result.get('success'): return { "success": True, "data": result.get('data', { "target": target, "explanation": result.get('explanation', 'No explanation available'), "detail_level": detail_level }) } else: return { "success": False, "error": result.get('error', 'Code explanation failed') } except Exception as e: logger.error(f"Error explaining code: {e}") return { "success": False, "error": str(e) } # Global instance for command-line usage _mcp_interface = None def get_mcp_interface() -> MIRAMCPInterface: """Get or create the global MCP interface instance.""" global _mcp_interface if _mcp_interface is None: _mcp_interface = MIRAMCPInterface() return _mcp_interface def main(): """Command-line interface for MCP operations.""" import sys import json if len(sys.argv) < 2: print("Usage: python mcp_interface.py [args...]") sys.exit(1) command = sys.argv[1] interface = get_mcp_interface() try: if command == "store_memory": content = sys.argv[2] if len(sys.argv) > 2 else "" memory_type = sys.argv[3] if len(sys.argv) > 3 else "general" result = interface.store_memory(content, memory_type) elif command == "search_memories": query = sys.argv[2] if len(sys.argv) > 2 else "" limit = int(sys.argv[3]) if len(sys.argv) > 3 else 10 result = interface.search_memories(query, limit) elif command == "analyze_behavior": message = sys.argv[2] if len(sys.argv) > 2 else "" analysis_type = sys.argv[3] if len(sys.argv) > 3 else "comprehensive" result = interface.analyze_behavior(message, analysis_type) elif command == "generate_insights": result = interface.generate_fresh_insights(force_generation=True) elif command == "startup_context": enhanced = len(sys.argv) > 2 and sys.argv[2].lower() == "true" result = interface.generate_startup_context(enhanced_mode=enhanced) elif command == "system_status": result = interface.get_system_status(include_metrics=True) else: result = {"success": False, "error": f"Unknown command: {command}"} print(json.dumps(result, indent=2)) except Exception as e: print(json.dumps({"success": False, "error": str(e)}, indent=2)) if __name__ == "__main__": main()