Spaces:
Sleeping
Sleeping
| """ | |
| 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() | |