client.py

"""
MCP client for interacting with the TutorX MCP server.
This module provides functions that interact with the MCP server
for use by the Gradio interface.
"""

import json
import aiohttp
import asyncio
from typing import Dict, Any, List, Optional
import base64
from datetime import datetime
import os

# Get server configuration from environment variables with defaults
DEFAULT_HOST = os.getenv("MCP_HOST", "127.0.0.1")
DEFAULT_PORT = int(os.getenv("MCP_PORT", "8000"))
DEFAULT_SERVER_URL = f"http://{DEFAULT_HOST}:{DEFAULT_PORT}"

class TutorXClient:
    """Client for interacting with the TutorX MCP server"""
    
    def __init__(self, server_url=DEFAULT_SERVER_URL):
        self.server_url = server_url
        self.session = None
    
    async def _ensure_session(self):
        """Ensure aiohttp session exists"""
        if self.session is None:
            self.session = aiohttp.ClientSession(
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json"
                }
            )
    
    async def _call_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        Call an MCP tool on the server
        
        Args:
            tool_name: Name of the tool to call
            params: Parameters to pass to the tool
            
        Returns:
            Tool response
        """
        await self._ensure_session()
        try:
            async with self.session.post(
                f"{self.server_url}/mcp/tools/{tool_name}",
                json=params,
                timeout=30
            ) as response:
                response.raise_for_status()
                return await response.json()
        except Exception as e:
            return {
                "error": f"Failed to call tool: {str(e)}",
                "timestamp": datetime.now().isoformat()
            }
    
    async def _get_resource(self, resource_uri: str) -> Dict[str, Any]:
        """
        Get an MCP resource from the server
        
        Args:
            resource_uri: URI of the resource to get
            
        Returns:
            Resource data
        """
        await self._ensure_session()
        try:
            async with self.session.get(
                f"{self.server_url}/mcp/resources?uri={resource_uri}",
                timeout=30
            ) as response:
                response.raise_for_status()
                return await response.json()
        except Exception as e:
            return {
                "error": f"Failed to get resource: {str(e)}",
                "timestamp": datetime.now().isoformat()
            }
    
    async def check_server_connection(self) -> bool:
        """
        Check if the server is accessible
        
        Returns:
            bool: True if server is accessible, False otherwise
        """
        await self._ensure_session()
        try:
            async with self.session.get(
                f"{self.server_url}/health",
                timeout=5
            ) as response:
                return response.status == 200
        except:
            return False
    
    # ------------ Core Features ------------
    
    async def assess_skill(self, student_id: str, concept_id: str) -> Dict[str, Any]:
        """Assess student's skill level on a specific concept"""
        return await self._call_tool("assess_skill", {
            "student_id": student_id,
            "concept_id": concept_id
        })
    
    async def get_concept_graph(self) -> Dict[str, Any]:
        """Get the full knowledge concept graph"""
        return await self._get_resource("concept-graph://")
    
    async def get_learning_path(self, student_id: str) -> Dict[str, Any]:
        """Get personalized learning path for a student"""
        return await self._get_resource(f"learning-path://{student_id}")
    
    async def generate_quiz(self, concept_ids: List[str], difficulty: int = 2) -> Dict[str, Any]:
        """Generate a quiz based on specified concepts and difficulty"""
        return await self._call_tool("generate_quiz", {
            "concept_ids": concept_ids,
            "difficulty": difficulty
        })
    
    async def analyze_error_patterns(self, student_id: str, concept_id: str) -> Dict[str, Any]:
        """Analyze common error patterns for a student on a specific concept"""
        return await self._call_tool("analyze_error_patterns", {
            "student_id": student_id,
            "concept_id": concept_id
        })
    
    # ------------ Advanced Features ------------
    
    async def analyze_cognitive_state(self, eeg_data: Dict[str, Any]) -> Dict[str, Any]:
        """Analyze EEG data to determine cognitive state"""
        return await self._call_tool("analyze_cognitive_state", {
            "eeg_data": eeg_data
        })
    
    async def get_curriculum_standards(self, country_code: str) -> Dict[str, Any]:
        """Get curriculum standards for a specific country"""
        return await self._get_resource(f"curriculum-standards://{country_code}")
    
    async def align_content_to_standard(self, content_id: str, standard_id: str) -> Dict[str, Any]:
        """Align educational content to a specific curriculum standard"""
        return await self._call_tool("align_content_to_standard", {
            "content_id": content_id,
            "standard_id": standard_id
        })
    
    async def generate_lesson(self, topic: str, grade_level: int, duration_minutes: int = 45) -> Dict[str, Any]:
        """Generate a complete lesson plan on a topic"""
        return await self._call_tool("generate_lesson", {
            "topic": topic,
            "grade_level": grade_level,
            "duration_minutes": duration_minutes
        })
    
    # ------------ User Experience ------------
    
    async def get_student_dashboard(self, student_id: str) -> Dict[str, Any]:
        """Get dashboard data for a specific student"""
        return await self._get_resource(f"student-dashboard://{student_id}")
    
    async def get_accessibility_settings(self, student_id: str) -> Dict[str, Any]:
        """Get accessibility settings for a student"""
        return await self._call_tool("get_accessibility_settings", {
            "student_id": student_id
        })
    
    async def update_accessibility_settings(self, student_id: str, settings: Dict[str, Any]) -> Dict[str, Any]:
        """Update accessibility settings for a student"""
        return await self._call_tool("update_accessibility_settings", {
            "student_id": student_id,
            "settings": settings
        })
    
    # ------------ Multi-Modal Interaction ------------
    
    async def text_interaction(self, query: str, student_id: str) -> Dict[str, Any]:
        """Process a text query from the student"""
        return await self._call_tool("text_interaction", {
            "query": query,
            "student_id": student_id
        })
    
    async def voice_interaction(self, audio_data_base64: str, student_id: str) -> Dict[str, Any]:
        """Process voice input from the student"""
        return await self._call_tool("voice_interaction", {
            "audio_data_base64": audio_data_base64,
            "student_id": student_id
        })
    
    async def handwriting_recognition(self, image_data_base64: str, student_id: str) -> Dict[str, Any]:
        """Process handwritten input from the student"""
        return await self._call_tool("handwriting_recognition", {
            "image_data_base64": image_data_base64,
            "student_id": student_id
        })
    
    # ------------ Assessment ------------
    
    async def create_assessment(self, concept_ids: List[str], num_questions: int, difficulty: int = 3) -> Dict[str, Any]:
        """Create a complete assessment for given concepts"""
        return await self._call_tool("create_assessment", {
            "concept_ids": concept_ids,
            "num_questions": num_questions,
            "difficulty": difficulty
        })
    
    async def grade_assessment(self, assessment_id: str, student_answers: Dict[str, str], questions: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Grade a completed assessment"""
        return await self._call_tool("grade_assessment", {
            "assessment_id": assessment_id,
            "student_answers": student_answers,
            "questions": questions
        })
    
    async def get_student_analytics(self, student_id: str, timeframe_days: int = 30) -> Dict[str, Any]:
        """Get comprehensive analytics for a student"""
        return await self._call_tool("get_student_analytics", {
            "student_id": student_id,
            "timeframe_days": timeframe_days
        })
    
    async def check_submission_originality(self, submission: str, reference_sources: List[str]) -> Dict[str, Any]:
        """Check student submission for potential plagiarism"""
        return await self._call_tool("check_submission_originality", {
            "submission": submission,
            "reference_sources": reference_sources
        })
    
    async def close(self):
        """Close the aiohttp session"""
        if self.session:
            await self.session.close()
            self.session = None

# Create a default client instance for easy import
client = TutorXClient()