modules/mcp_integration.py

# type: ignore
"""
================================================================================
MCP INTEGRATION MODULE - ANTHROPIC MODEL CONTEXT PROTOCOL
================================================================================
Integrates MCP (Model Context Protocol) as optional skill provider for AKIRA.
MCP expands available tools and resources without breaking existing architecture.

Philosophy:
- MCP is OPTIONAL (fallback to current skills if unavailable)
- MCP tools augment existing skills, don't replace them
- Zero impact on current behavior if MCP is disabled
- Lazy loading - only initialize if requested

Supported MCP Resources:
- filesystem_read: Safe file reading (whitelist)
- filesystem_write: Safe file writing (whitelist)
- web_search_advanced: Enhanced web search
- database_query: Query AKIRA memory DB
- skill_execute: Execute existing AKIRA skills
- system_info: System statistics
================================================================================
"""

import os
import json
import logging
import asyncio
from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime

try:
    from loguru import logger
except ImportError:
    logger = logging.getLogger(__name__)

# ============================================================
# MCP RESOURCE DEFINITIONS
# ============================================================

@dataclass
class MCPResourceSchema:
    """Schema for an MCP resource (tool)."""
    name: str
    description: str
    parameters: Dict[str, Any]
    handler: Optional[Callable] = None
    enabled: bool = True
    category: str = "general"  # filesystem, web, database, skill, system


# ============================================================
# MCP RESOURCE CATALOG
# ============================================================

class MCPResourceCatalog:
    """
    Central catalog of available MCP resources.
    Defines all tools that MCP can expose to Claude.
    """
    
    def __init__(self):
        self.logger = logger
        self.resources: Dict[str, MCPResourceSchema] = {}
        self._initialize_default_resources()
    
    def _initialize_default_resources(self):
        """Initialize built-in MCP resources."""
        
        # FILESYSTEM_READ
        self.resources["filesystem_read"] = MCPResourceSchema(
            name="filesystem_read",
            description="Read content from a file on the system (whitelist only)",
            parameters={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "File path to read (must be whitelisted)"
                    },
                    "encoding": {
                        "type": "string",
                        "description": "File encoding (default: utf-8)",
                        "default": "utf-8"
                    }
                },
                "required": ["path"]
            },
            category="filesystem"
        )
        
        # FILESYSTEM_WRITE
        self.resources["filesystem_write"] = MCPResourceSchema(
            name="filesystem_write",
            description="Write content to a file (whitelisted directories only)",
            parameters={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "File path to write to (must be whitelisted)"
                    },
                    "content": {
                        "type": "string",
                        "description": "Content to write"
                    }
                },
                "required": ["path", "content"]
            },
            category="filesystem"
        )
        
        # WEB_SEARCH_ADVANCED
        self.resources["web_search_advanced"] = MCPResourceSchema(
            name="web_search_advanced",
            description="Perform advanced web search with filters and ranking",
            parameters={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search query"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum results to return (default: 5)",
                        "default": 5
                    },
                    "filter": {
                        "type": "string",
                        "description": "Optional filter (news, academic, recent)",
                        "enum": ["news", "academic", "recent", "none"]
                    }
                },
                "required": ["query"]
            },
            category="web"
        )
        
        # DATABASE_QUERY
        self.resources["database_query"] = MCPResourceSchema(
            name="database_query",
            description="Query AKIRA's memory database for facts and patterns",
            parameters={
                "type": "object",
                "properties": {
                    "user_id": {
                        "type": "string",
                        "description": "User ID to query"
                    },
                    "query_type": {
                        "type": "string",
                        "description": "Type of query (facts, patterns, emotions, preferences)",
                        "enum": ["facts", "patterns", "emotions", "preferences"]
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Number of results (default: 10)",
                        "default": 10
                    }
                },
                "required": ["user_id", "query_type"]
            },
            category="database"
        )
        
        # SYSTEM_INFO
        self.resources["system_info"] = MCPResourceSchema(
            name="system_info",
            description="Get current system information (time, date, stats)",
            parameters={
                "type": "object",
                "properties": {
                    "info_type": {
                        "type": "string",
                        "description": "What info to retrieve",
                        "enum": ["time", "date", "datetime", "uptime", "memory", "all"]
                    }
                },
                "required": ["info_type"]
            },
            category="system"
        )
        
        # SKILL_EXECUTE
        self.resources["skill_execute"] = MCPResourceSchema(
            name="skill_execute",
            description="Execute an existing AKIRA skill",
            parameters={
                "type": "object",
                "properties": {
                    "skill_name": {
                        "type": "string",
                        "description": "Name of the skill to execute"
                    },
                    "arguments": {
                        "type": "object",
                        "description": "Arguments to pass to the skill"
                    }
                },
                "required": ["skill_name"]
            },
            category="skill"
        )
        
        self.logger.info(f"✅ MCP Catalog initialized with {len(self.resources)} resources")
    
    def get_resource(self, name: str) -> Optional[MCPResourceSchema]:
        """Get a specific resource schema."""
        return self.resources.get(name)
    
    def list_resources(self, category: Optional[str] = None) -> List[MCPResourceSchema]:
        """List all available resources, optionally filtered by category."""
        if category:
            return [r for r in self.resources.values() if r.category == category]
        return list(self.resources.values())
    
    def to_claude_tools_format(self) -> List[Dict[str, Any]]:
        """
        Convert resources to Claude Tool Use format.
        Returns list of tool definitions compatible with Claude API.
        """
        tools = []
        for resource in self.resources.values():
            if not resource.enabled:
                continue
            
            tool = {
                "name": resource.name,
                "description": resource.description,
                "input_schema": {
                    "type": "object",
                    "properties": resource.parameters.get("properties", {}),
                    "required": resource.parameters.get("required", [])
                }
            }
            tools.append(tool)
        
        return tools


# ============================================================
# MCP SERVER CLIENT
# ============================================================

class MCPServerClient:
    """
    Client for communicating with MCP server(s).
    Handles resource invocation, caching, and error handling.
    """
    
    def __init__(self, catalog: Optional[MCPResourceCatalog] = None):
        self.logger = logger
        self.catalog = catalog or MCPResourceCatalog()
        self.is_available = False
        self._detection_attempted = False
        self._default_handlers = self._setup_default_handlers()
        self._detect_mcp_server()
    
    def _setup_default_handlers(self) -> Dict[str, Callable]:
        """Setup default handlers for resources."""
        return {
            "system_info": self._handle_system_info,
            "filesystem_read": self._handle_filesystem_read,
            "filesystem_write": self._handle_filesystem_write,
            "skill_execute": self._handle_skill_execute,
        }

    
    def _detect_mcp_server(self):
        """Detect if MCP server is available."""
        try:
            # Check for common MCP server indicators
            # For now, assume available if module can be imported
            import anthropic
            self.is_available = True
            self.logger.info("✅ MCP Server detected (anthropic module available)")
        except ImportError:
            self.is_available = False
            self.logger.warning("⚠️ Anthropic SDK not available - MCP disabled")
        except Exception as e:
            self.is_available = False
            self.logger.warning(f"⚠️ MCP detection failed: {e}")
        
        self._detection_attempted = True
    
    def _handle_system_info(self, info_type: str = "all") -> Dict[str, Any]:
        """Handle system_info resource."""
        import psutil
        from datetime import datetime
        
        result = {}
        
        if info_type in ["time", "all"]:
            result["time"] = datetime.now().strftime("%H:%M:%S")
        
        if info_type in ["date", "all"]:
            result["date"] = datetime.now().strftime("%Y-%m-%d")
        
        if info_type in ["datetime", "all"]:
            result["datetime"] = datetime.now().isoformat()
        
        if info_type in ["uptime", "all"]:
            try:
                uptime = psutil.boot_time()
                result["uptime_seconds"] = int(datetime.now().timestamp() - uptime)
            except:
                pass
        
        if info_type in ["memory", "all"]:
            try:
                memory = psutil.virtual_memory()
                result["memory_percent"] = memory.percent
                result["memory_available_gb"] = memory.available / (1024**3)
            except:
                pass
        
        return result
    
    def _handle_filesystem_read(self, path: str, encoding: str = "utf-8") -> Dict[str, Any]:
        """Handle filesystem_read resource (safe paths only)."""
        # Whitelist of safe directories
        safe_dirs = [
            "/akira/data",
            "/akira/logs",
            os.path.expanduser("~/.akira"),
        ]
        
        # Check if path is in whitelist
        path_allowed = any(path.startswith(safe_dir) for safe_dir in safe_dirs)
        if not path_allowed:
            return {"error": "Path not whitelisted for reading"}
        
        try:
            with open(path, 'r', encoding=encoding) as f:
                content = f.read()
            return {"success": True, "content": content}
        except Exception as e:
            return {"error": str(e)}
    
    def _handle_filesystem_write(self, path: str, content: str) -> Dict[str, Any]:
        """Handle filesystem_write resource (safe paths only)."""
        # Whitelist of safe directories
        safe_dirs = [
            "/akira/data",
            os.path.expanduser("~/.akira"),
        ]
        
        # Check if path is in whitelist
        path_allowed = any(path.startswith(safe_dir) for safe_dir in safe_dirs)
        if not path_allowed:
            return {"error": "Path not whitelisted for writing"}
        
        try:
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, 'w', encoding='utf-8') as f:
                f.write(content)
            return {"success": True, "message": f"Written to {path}"}
        except Exception as e:
            return {"error": str(e)}
    
    def _handle_skill_execute(self, skill_name: str, arguments: Dict[str, Any] = None) -> Dict[str, Any]:
        """
        ✅ NOVO: Handle skill_execute resource
        Executa qualquer skill registrada no AKIRA registry
        """
        try:
            from modules.skills_registry import registry
            
            if arguments is None:
                arguments = {}
            
            self.logger.info(f"🛠️ [MCP] Executando skill: {skill_name} com args: {arguments}")
            
            result = registry.execute(skill_name, arguments)
            
            return {
                "success": True,
                "skill": skill_name,
                "result": result,
                "timestamp": datetime.now().isoformat()
            }
        except ImportError:
            return {
                "success": False,
                "error": "Skills registry not available",
                "skill": skill_name
            }
        except Exception as e:
            self.logger.error(f"❌ [MCP] Erro ao executar skill {skill_name}: {e}")
            return {
                "success": False,
                "error": str(e),
                "skill": skill_name
            }
    
    async def invoke_resource(self, resource_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """
        Invoke an MCP resource.
        Tries handler first, then falls back to error.
        """
        if not self.is_available:
            return {"error": "MCP server not available"}
        
        handler = self._default_handlers.get(resource_name)
        if handler:
            try:
                result = handler(**arguments)
                return result
            except Exception as e:
                return {"error": str(e)}
        
        return {"error": f"Resource '{resource_name}' handler not found"}
    
    def get_available_tools(self) -> List[Dict[str, Any]]:
        """Get all available tools in Claude format."""
        if not self.is_available:
            return []
        return self.catalog.to_claude_tools_format()


# ============================================================
# SINGLETON INSTANCE
# ============================================================

_MCP_CATALOG = None
_MCP_CLIENT = None

def get_mcp_catalog() -> MCPResourceCatalog:
    """Get singleton MCP catalog."""
    global _MCP_CATALOG
    if _MCP_CATALOG is None:
        _MCP_CATALOG = MCPResourceCatalog()
    return _MCP_CATALOG

def get_mcp_client() -> MCPServerClient:
    """Get singleton MCP client."""
    global _MCP_CLIENT
    if _MCP_CLIENT is None:
        _MCP_CLIENT = MCPServerClient(get_mcp_catalog())
    return _MCP_CLIENT