""" 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", "8001")) # Default port updated to 8001 DEFAULT_SERVER_URL = f"http://{DEFAULT_HOST}:{DEFAULT_PORT}" # API endpoints API_PREFIX = "/api" 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], method: str = "POST") -> 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 method: HTTP method to use (GET or POST) Returns: Tool response """ await self._ensure_session() try: url = f"{self.server_url}{API_PREFIX}/{tool_name}" # Convert params to query string for GET requests if method.upper() == "GET": from urllib.parse import urlencode if params: query_string = urlencode(params, doseq=True) url = f"{url}?{query_string}" async with self.session.get(url, timeout=30) as response: return await self._handle_response(response) else: async with self.session.post(url, json=params, timeout=30) as response: return await self._handle_response(response) except Exception as e: return { "error": f"Failed to call tool: {str(e)}", "timestamp": datetime.now().isoformat() } async def _handle_response(self, response) -> Dict[str, Any]: """Handle the HTTP response""" if response.status == 200: return await response.json() else: error = await response.text() return { "error": f"API error ({response.status}): {error}", "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: # Extract the resource name from the URI (e.g., 'concept-graph' from 'concept-graph://') resource_name = resource_uri.split('://')[0] url = f"{self.server_url}{API_PREFIX}/{resource_name}" async with self.session.get(url, timeout=30) as response: if response.status == 200: return await response.json() else: error = await response.text() return { "error": f"Failed to get resource: {error}", "timestamp": datetime.now().isoformat() } 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}{API_PREFIX}/health", timeout=5 ) as response: return response.status == 200 except Exception as e: print(f"Server connection check failed: {str(e)}") return False # ------------ Core Features ------------ async def get_concept_graph(self, concept_id: str = None, use_mcp: bool = False) -> Dict[str, Any]: """ Get the concept graph for a specific concept or all concepts. Args: concept_id: Optional ID of the concept to fetch. If None, returns all concepts. use_mcp: If True, uses the MCP tool interface instead of direct API call. Returns: Dict containing concept data or error information. """ try: # Ensure we have a session await self._ensure_session() if use_mcp: # Use MCP tool interface print(f"[CLIENT] Using MCP tool to get concept graph for: {concept_id}") return await self._call_tool("get_concept_graph", {"concept_id": concept_id} if concept_id else {}) # Use direct API call (default) url = f"{self.server_url}/api/concept_graph" params = {} if concept_id: params["concept_id"] = concept_id print(f"[CLIENT] Fetching concept graph from {url} with params: {params}") async with self.session.get( url, params=params, timeout=30 ) as response: print(f"[CLIENT] Response status: {response.status}") if response.status == 404: error_msg = f"Concept {concept_id} not found" print(f"[CLIENT] {error_msg}") return {"error": error_msg} response.raise_for_status() # Parse the JSON response result = await response.json() print(f"[CLIENT] Received response: {result}") return result except asyncio.TimeoutError: error_msg = "Request timed out" print(f"[CLIENT] {error_msg}") return {"error": error_msg} except aiohttp.ClientError as e: error_msg = f"HTTP client error: {str(e)}" print(f"[CLIENT] {error_msg}") return {"error": error_msg} except Exception as e: error_msg = f"Unexpected error: {str(e)}" print(f"[CLIENT] {error_msg}") import traceback traceback.print_exc() return {"error": error_msg} async def assess_skill(self, student_id: str, concept_id: str) -> Dict[str, Any]: """Assess a student's skill on a specific concept""" return await self._call_tool("assess_skill", {"student_id": student_id, "concept_id": concept_id}) 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 get_curriculum_standards(self, country_code: str = "us") -> Dict[str, Any]: """ Get curriculum standards for a specific country Args: country_code: ISO country code (e.g., 'us', 'uk') Returns: Dictionary containing curriculum standards """ return await self._call_tool( "curriculum-standards", # Note the endpoint name matches the API route {"country": country_code.lower()}, # Note the parameter name matches the API method="GET" # Use GET for this endpoint ) async def close(self): """Close the aiohttp session""" if self.session: await self.session.close() self.session = None async def generate_lesson(self, topic: str, grade_level: int, duration_minutes: int) -> Dict[str, Any]: """ Generate a lesson plan for the given topic, grade level, and duration Args: topic: The topic for the lesson grade_level: The grade level (1-12) duration_minutes: Duration of the lesson in minutes Returns: Dictionary containing the generated lesson plan """ return await self._call_tool( "generate_lesson", { "topic": topic, "grade_level": grade_level, "duration_minutes": duration_minutes } ) # Create a default client instance for easy import client = TutorXClient()