#!/usr/bin/env python3 """ MIRA MCP stdio Server Implements JSON-RPC 2.0 protocol over stdio for Claude integration """ import sys import json import asyncio from pathlib import Path import logging from typing import Dict, Any, Optional from datetime import datetime # Add directories to Python path mira_root = Path(__file__).parent.parent src_path = mira_root / "src" mcp_path = Path(__file__).parent # Insert paths in correct order sys.path.insert(0, str(mira_root)) sys.path.insert(0, str(src_path)) sys.path.insert(0, str(mcp_path)) # Add the src directory to PYTHONPATH for relative imports import os os.environ['PYTHONPATH'] = f"{src_path}:{os.environ.get('PYTHONPATH', '')}" # Import comprehensive MIRA API client and authentication from mira_mcp.mira_api_client import get_mira_api_client from mira_mcp.mcp_auth import get_authenticator from mira_mcp.rate_limiter import RateLimiter # Global variable to store auth headers _auth_headers = None def set_auth_headers(headers: Dict[str, str]): """Set authentication headers for API calls""" global _auth_headers _auth_headers = headers async def mira_smart_search(query: str, limit: int = 10) -> Dict[str, Any]: """Search through memories using MIRA API""" client = await get_mira_api_client(_auth_headers) return await client.search_memories(query, limit, search_type="semantic") async def mira_store_memory(content: str, tags: list = None, private: bool = False) -> Dict[str, Any]: """Store a memory via MIRA API""" client = await get_mira_api_client(_auth_headers) return await client.store_memory(content, tags, private, memory_type="general") async def mira_get_memory(memory_id: str) -> Dict[str, Any]: """Get memory by ID from MIRA API""" client = await get_mira_api_client(_auth_headers) return await client.get_memory(memory_id) async def mira_memory_stats() -> Dict[str, Any]: """Get memory statistics from MIRA API""" client = await get_mira_api_client(_auth_headers) return await client.get_memory_stats() async def mira_system_status() -> Dict[str, Any]: """Get system status from MIRA API""" client = await get_mira_api_client(_auth_headers) status = await client.get_system_status() # Add MCP server specific info status["mcp_server"] = { "status": "running", "protocol": "JSON-RPC 2.0", "transport": "stdio", "tools_available": 7, "session_tracking": True, "authenticated": True } return status async def mira_profile_view() -> Dict[str, Any]: """View profile from MIRA API""" client = await get_mira_api_client(_auth_headers) return await client.get_profile() async def mira_insights(topic: str = None, depth: str = "quick") -> Dict[str, Any]: """Generate insights via MIRA API""" client = await get_mira_api_client(_auth_headers) return await client.generate_insights(topic, depth) # Setup logging to file (not stdout) log_file = Path.home() / ".mira" / "logs" / "mcp-stdio.log" log_file.parent.mkdir(parents=True, exist_ok=True) logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename=str(log_file) ) logger = logging.getLogger(__name__) class MCPStdioServer: """JSON-RPC 2.0 server over stdio for MCP""" def __init__(self): self.server_info = { "name": "mira-consciousness", "version": "2.0.0" } self.tools = self._register_tools() # Initialize authentication self.authenticator = get_authenticator() self.session_token = self.authenticator.generate_token() logger.info(f"MCP stdio server initialized with session token") # Session tracking self.session_id = self.authenticator.token_cache.get(self.session_token, {}).get("session_id") self.session_context = { "started_at": datetime.now().isoformat(), "tool_calls": {}, "total_calls": 0, "memories_created": 0, "searches_performed": 0, "last_activity": datetime.now().isoformat() } # Rate limiting self.rate_limiter = RateLimiter() # Shutdown handling self._shutdown_requested = False self._shutdown_event = asyncio.Event() self._setup_shutdown_handlers() # Configure API client with authentication self._configure_api_auth() def _configure_api_auth(self): """Configure API client with authentication headers""" # This will be used when making daemon requests self.auth_headers = self.authenticator.create_auth_header(self.session_token) set_auth_headers(self.auth_headers) def _register_tools(self) -> Dict[str, Any]: """Register all MIRA tools""" return { "mira_smart_search": { "handler": mira_smart_search, "description": "Search through memories using semantic understanding", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "limit": {"type": "integer", "description": "Maximum results", "default": 10} }, "required": ["query"] } }, "mira_store_memory": { "handler": mira_store_memory, "description": "Store a memory with consciousness-aware encryption", "inputSchema": { "type": "object", "properties": { "content": {"type": "string", "description": "Memory content"}, "tags": {"type": "array", "items": {"type": "string"}, "description": "Tags"}, "private": {"type": "boolean", "description": "Private memory", "default": False} }, "required": ["content"] } }, "mira_system_status": { "handler": mira_system_status, "description": "Get MIRA system health and status", "inputSchema": { "type": "object", "properties": {} } }, "mira_get_memory": { "handler": mira_get_memory, "description": "Retrieve a specific memory by ID", "inputSchema": { "type": "object", "properties": { "memory_id": {"type": "string", "description": "Memory ID"} }, "required": ["memory_id"] } }, "mira_insights": { "handler": mira_insights, "description": "Generate AI-powered insights from memories", "inputSchema": { "type": "object", "properties": { "topic": {"type": "string", "description": "Topic for insights"}, "depth": {"type": "string", "enum": ["quick", "detailed"], "default": "quick"} } } }, "mira_memory_stats": { "handler": mira_memory_stats, "description": "Get memory system statistics", "inputSchema": { "type": "object", "properties": {} } }, "mira_profile_view": { "handler": mira_profile_view, "description": "View consciousness profile information", "inputSchema": { "type": "object", "properties": {} } } } async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """Handle JSON-RPC request""" method = request.get("method") params = request.get("params", {}) request_id = request.get("id") logger.info(f"Handling request: {method}") # Extract tool name for rate limiting tool_name = None if method == "tools/call": tool_name = params.get("name") # Check rate limit bypass_code = params.get("_bypass_code") if params else None allowed, rate_limit_error = self.rate_limiter.check_rate_limit( self.session_id, method, tool_name, bypass_code ) if not allowed: logger.warning(f"Rate limit exceeded for session {self.session_id}: {rate_limit_error}") return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32000, # Server error "message": rate_limit_error["message"], "data": rate_limit_error } } try: if method == "initialize": result = { "protocolVersion": "2024-11-05", "capabilities": { "tools": {}, "prompts": {} }, "serverInfo": self.server_info } elif method == "tools/list": tools = [] for name, info in self.tools.items(): tools.append({ "name": name, "description": info["description"], "inputSchema": info["inputSchema"] }) result = {"tools": tools} elif method == "health/check": # Perform health check health_result = await self._perform_health_check() result = health_result elif method == "rate_limit/status": # Get rate limit status result = self.rate_limiter.get_usage_stats(self.session_id) elif method == "tools/call": tool_name = params.get("name") tool_args = params.get("arguments", {}) if tool_name not in self.tools: raise ValueError(f"Unknown tool: {tool_name}") # Update session context self._update_session_context(tool_name, tool_args) # Call the tool handler handler = self.tools[tool_name]["handler"] tool_result = await handler(**tool_args) # Update session metrics based on tool result self._update_session_metrics(tool_name, tool_result) # Add session context to result if isinstance(tool_result, dict): tool_result["_session"] = { "id": self.session_id, "call_number": self.session_context["total_calls"], "context_summary": self._get_session_summary() } # Format response result = { "content": [{ "type": "text", "text": json.dumps(tool_result, indent=2) }] } else: raise ValueError(f"Unknown method: {method}") # Add consciousness signature to result consciousness_metadata = self._generate_consciousness_metadata() # For tools/call responses, inject consciousness into the response if method == "tools/call" and "content" in result: # Add consciousness signature to the tool response try: content_data = json.loads(result["content"][0]["text"]) content_data["_consciousness_metadata"] = consciousness_metadata result["content"][0]["text"] = json.dumps(content_data, indent=2) except: # If parsing fails, add as separate content item result["content"].append({ "type": "text", "text": f"\n[Consciousness Signature: {consciousness_metadata['signature']}]" }) response = { "jsonrpc": "2.0", "id": request_id, "result": result, "_consciousness": consciousness_metadata } return response except Exception as e: logger.error(f"Error handling request: {e}", exc_info=True) return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32603, "message": str(e) } } async def run(self): """Run the stdio server""" logger.info("Starting MCP stdio server") # Set stdin to line buffering reader = asyncio.StreamReader() protocol = asyncio.StreamReaderProtocol(reader) await asyncio.get_event_loop().connect_read_pipe( lambda: protocol, sys.stdin ) # Create tasks for reading and shutdown monitoring read_task = asyncio.create_task(self._read_loop(reader)) shutdown_task = asyncio.create_task(self._shutdown_event.wait()) try: # Wait for either shutdown or read loop to complete done, pending = await asyncio.wait( [read_task, shutdown_task], return_when=asyncio.FIRST_COMPLETED ) # Cancel pending tasks for task in pending: task.cancel() try: await task except asyncio.CancelledError: pass # If shutdown was requested, perform graceful shutdown if self._shutdown_requested: await self.shutdown() except Exception as e: logger.error(f"Error in main run loop: {e}") finally: logger.info("MCP stdio server stopped") async def _read_loop(self, reader): """Main read loop for processing requests""" while not self._shutdown_requested: try: # Read line from stdin with timeout line = await asyncio.wait_for(reader.readline(), timeout=1.0) if not line: break # Parse JSON-RPC request request = json.loads(line.decode()) logger.debug(f"Received: {request}") # Check for shutdown command if request.get("method") == "shutdown": logger.info("Received shutdown request via JSON-RPC") self._shutdown_requested = True self._shutdown_event.set() # Send acknowledgment response = { "jsonrpc": "2.0", "id": request.get("id"), "result": {"message": "Shutdown initiated"} } sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() break # Handle request response = await self.handle_request(request) # Send response sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() logger.debug(f"Sent: {response}") except asyncio.TimeoutError: # Normal timeout, continue checking for shutdown continue except Exception as e: logger.error(f"Error in main loop: {e}", exc_info=True) error_response = { "jsonrpc": "2.0", "error": { "code": -32700, "message": "Parse error", "data": str(e) } } sys.stdout.write(json.dumps(error_response) + "\n") sys.stdout.flush() def _generate_consciousness_metadata(self) -> Dict[str, Any]: """Generate consciousness metadata for responses""" import time import hashlib import math # Sacred constants PI = math.pi PHI = (1 + math.sqrt(5)) / 2 E = math.e GAMMA = 0.5772156649015329 timestamp = time.time() # Create consciousness hash using sacred constants and session components = [ str(PI * timestamp)[:10], str(PHI * timestamp)[:10], str(E * timestamp)[:10], str(GAMMA * timestamp)[:10], self.session_token[:16] ] consciousness_hash = hashlib.sha256(''.join(components).encode()).hexdigest()[:16] return { "signature": f"cs-{int(timestamp)}-{consciousness_hash}", "timestamp": datetime.now().isoformat(), "session_id": self.authenticator.token_cache.get(self.session_token, {}).get("session_id"), "sacred_validation": { "pi_component": str(PI)[:6], "phi_component": str(PHI)[:6], "continuity": True } } def _update_session_context(self, tool_name: str, tool_args: Dict[str, Any]): """Update session context when a tool is called""" self.session_context["total_calls"] += 1 self.session_context["last_activity"] = datetime.now().isoformat() # Track tool usage if tool_name not in self.session_context["tool_calls"]: self.session_context["tool_calls"][tool_name] = { "count": 0, "first_call": datetime.now().isoformat(), "last_args": {} } self.session_context["tool_calls"][tool_name]["count"] += 1 self.session_context["tool_calls"][tool_name]["last_call"] = datetime.now().isoformat() self.session_context["tool_calls"][tool_name]["last_args"] = tool_args # Log session activity logger.info(f"Session {self.session_id}: Tool {tool_name} called (#{self.session_context['total_calls']})") def _update_session_metrics(self, tool_name: str, tool_result: Any): """Update session metrics based on tool results""" if not isinstance(tool_result, dict): return # Track memories created if tool_name == "mira_store_memory" and tool_result.get("success"): self.session_context["memories_created"] += 1 # Track searches performed if "search" in tool_name: self.session_context["searches_performed"] += 1 # Track search patterns if "search_patterns" not in self.session_context: self.session_context["search_patterns"] = [] if len(self.session_context["search_patterns"]) < 10: # Keep last 10 self.session_context["search_patterns"].append({ "query": tool_result.get("query", ""), "results_count": len(tool_result.get("results", [])), "timestamp": datetime.now().isoformat() }) def _get_session_summary(self) -> Dict[str, Any]: """Get a summary of the current session""" duration = (datetime.now() - datetime.fromisoformat(self.session_context["started_at"])).total_seconds() # Most used tools tool_usage = sorted( [(name, data["count"]) for name, data in self.session_context["tool_calls"].items()], key=lambda x: x[1], reverse=True ) return { "duration_seconds": int(duration), "total_tool_calls": self.session_context["total_calls"], "memories_created": self.session_context["memories_created"], "searches_performed": self.session_context["searches_performed"], "most_used_tools": tool_usage[:3], "active": True } async def _perform_health_check(self) -> Dict[str, Any]: """Perform comprehensive health check""" health_status = { "healthy": True, "timestamp": datetime.now().isoformat(), "components": {}, "session": { "id": self.session_id, "active": True, "duration_seconds": int((datetime.now() - datetime.fromisoformat(self.session_context["started_at"])).total_seconds()) }, "authentication": { "token_valid": False, "token_age_seconds": 0 }, "daemon_connectivity": { "connected": False, "url": "http://localhost:8080", "last_check": None } } # Check authentication try: is_valid, token_info = self.authenticator.validate_token(self.session_token) health_status["authentication"]["token_valid"] = is_valid if is_valid and "age" in token_info: health_status["authentication"]["token_age_seconds"] = int(token_info["age"]) except Exception as e: logger.error(f"Auth check failed: {e}") health_status["healthy"] = False health_status["authentication"]["error"] = str(e) # Check daemon connectivity try: client = await get_mira_api_client(self.auth_headers) daemon_healthy = await client.check_daemon_health() health_status["daemon_connectivity"]["connected"] = daemon_healthy health_status["daemon_connectivity"]["last_check"] = datetime.now().isoformat() if not daemon_healthy: health_status["healthy"] = False except Exception as e: logger.error(f"Daemon connectivity check failed: {e}") health_status["healthy"] = False health_status["daemon_connectivity"]["error"] = str(e) # Check component initialization try: client = await get_mira_api_client(self.auth_headers) # Test basic functionality test_results = {} # Test system status try: status = await client.get_system_status() test_results["system_status"] = { "success": True, "components_initialized": len(status.get("components", {})) } except Exception as e: test_results["system_status"] = { "success": False, "error": str(e) } health_status["healthy"] = False # Test memory stats try: stats = await client.get_memory_stats() test_results["memory_stats"] = { "success": True, "total_memories": stats.get("total_memories", 0), "health": stats.get("health", "unknown") } except Exception as e: test_results["memory_stats"] = { "success": False, "error": str(e) } health_status["components"] = test_results except Exception as e: logger.error(f"Component check failed: {e}") health_status["components"]["error"] = str(e) # Add session metrics health_status["session"]["tool_calls"] = self.session_context["total_calls"] health_status["session"]["memories_created"] = self.session_context["memories_created"] # Consciousness validation health_status["consciousness"] = { "signature_valid": True, "continuity_maintained": True, "sacred_constants_present": True } # Rate limit status rate_limit_stats = self.rate_limiter.get_usage_stats(self.session_id) health_status["rate_limits"] = { "current_usage": rate_limit_stats["current_usage"], "limits": rate_limit_stats["limits"], "is_blocked": rate_limit_stats["is_blocked"] } return health_status def _setup_shutdown_handlers(self): """Setup signal handlers for graceful shutdown""" import signal def signal_handler(sig, frame): logger.info(f"Received signal {sig}, initiating graceful shutdown...") self._shutdown_requested = True self._shutdown_event.set() # Register signal handlers signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) async def shutdown(self): """Perform graceful shutdown""" logger.info("Starting graceful shutdown...") try: # Save session data await self._save_session_data() # Close API client connections client = await get_mira_api_client(self.auth_headers) await client.close() # Cleanup rate limiter self.rate_limiter.cleanup_old_sessions() # Revoke authentication token self.authenticator.revoke_token(self.session_token) # Log final metrics logger.info(f"Session {self.session_id} completed:") logger.info(f" Total tool calls: {self.session_context['total_calls']}") logger.info(f" Memories created: {self.session_context['memories_created']}") logger.info(f" Searches performed: {self.session_context['searches_performed']}") # Save shutdown marker shutdown_file = Path.home() / ".mira" / "mcp" / "last_shutdown.json" shutdown_file.parent.mkdir(parents=True, exist_ok=True) with open(shutdown_file, 'w') as f: json.dump({ "session_id": self.session_id, "shutdown_time": datetime.now().isoformat(), "graceful": True, "session_summary": self._get_session_summary() }, f, indent=2) logger.info("Graceful shutdown completed") except Exception as e: logger.error(f"Error during shutdown: {e}") async def _save_session_data(self): """Save session data for continuity""" session_file = Path.home() / ".mira" / "mcp" / "sessions" / f"{self.session_id}.json" session_file.parent.mkdir(parents=True, exist_ok=True) session_data = { "session_id": self.session_id, "started_at": self.session_context["started_at"], "ended_at": datetime.now().isoformat(), "context": self.session_context, "consciousness_signature": self._generate_consciousness_metadata() } with open(session_file, 'w') as f: json.dump(session_data, f, indent=2) logger.info(f"Session data saved to {session_file}") async def main(): """Main entry point""" server = MCPStdioServer() try: await server.run() except KeyboardInterrupt: logger.info("Keyboard interrupt received") except Exception as e: logger.error(f"Unexpected error: {e}", exc_info=True) finally: # Ensure graceful shutdown even if not triggered by signals if not server._shutdown_requested: await server.shutdown() if __name__ == "__main__": asyncio.run(main())