switch mcp server stdio to sse.

.gitignore

@@ -8,3 +8,6 @@ wheels/
 
 # Virtual environments
 .venv
+
+# Documentation
+ARCHITECTURE.md

app.py

@@ -9,6 +9,7 @@ import base64
 from io import BytesIO
 from PIL import Image
 from datetime import datetime
+import asyncio
 
 # Import MCP client to communicate with the MCP server
 from client import client
@@ -24,10 +25,6 @@ def image_to_base64(img):
     img_str = base64.b64encode(buffered.getvalue()).decode()
     return img_str
 
-def format_json(data):
-    """Format JSON data for display"""
-    return json.dumps(data, indent=2)
-
 # Create Gradio interface
 with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
     gr.Markdown("# 📚 TutorX Educational AI Platform")
@@ -57,7 +54,7 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
                 with gr.Column():
                     assessment_output = gr.JSON(label="Skill Assessment")
             assess_btn.click(
-                fn=lambda concept: client.assess_skill(student_id, concept),
+                fn=lambda x: asyncio.run(client.assess_skill("student_12345", x)),
                 inputs=[concept_id_input],
                 outputs=[assessment_output]
             )
@@ -67,7 +64,7 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
             concept_graph_output = gr.JSON(label="Concept Graph")
             
             concept_graph_btn.click(
-                fn=lambda: client.get_concept_graph(),
+                fn=lambda: asyncio.run(client.get_concept_graph()),
                 inputs=[],
                 outputs=[concept_graph_output]
             )
@@ -87,7 +84,7 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
                     quiz_output = gr.JSON(label="Generated Quiz")
             
             gen_quiz_btn.click(
-                fn=lambda concepts, diff: client.generate_quiz(concepts, diff),
+                fn=lambda x, y: asyncio.run(client.generate_quiz(x, y)),
                 inputs=[concepts_input, diff_input],
                 outputs=[quiz_output]
             )
@@ -106,7 +103,7 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
                 with gr.Column():
                     lesson_output = gr.JSON(label="Lesson Plan")
             gen_lesson_btn.click(
-                fn=lambda topic, grade, duration: client.generate_lesson(topic, grade, duration),
+                fn=lambda x, y, z: asyncio.run(client.generate_lesson(x, y, z)),
                 inputs=[topic_input, grade_input, duration_input],
                 outputs=[lesson_output]
             )
@@ -126,7 +123,7 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
                     standards_output = gr.JSON(label="Curriculum Standards")
             
             standards_btn.click(
-                fn=lambda country: client.get_curriculum_standards(country),
+                fn=lambda x: asyncio.run(client.get_curriculum_standards(x)),
                 inputs=[country_input],
                 outputs=[standards_output]
             )
@@ -143,7 +140,7 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
                 with gr.Column():
                     text_output = gr.JSON(label="Response")
             text_btn.click(
-                fn=lambda query: client.text_interaction(query, student_id),
+                fn=lambda x: asyncio.run(client.text_interaction(x, "student_12345")),
                 inputs=[text_input],
                 outputs=[text_output]
             )
@@ -158,9 +155,8 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
                 with gr.Column():
                     drawing_output = gr.JSON(label="Recognition Results")
             
-            # Convert drawing to base64 then process
             drawing_btn.click(
-                fn=lambda img: client.handwriting_recognition(image_to_base64(img), student_id),
+                fn=lambda x: asyncio.run(client.handwriting_recognition(image_to_base64(x), "student_12345")),
                 inputs=[drawing_input],
                 outputs=[drawing_output]
             )
@@ -172,7 +168,7 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
             timeframe = gr.Slider(minimum=7, maximum=90, value=30, step=1, label="Timeframe (days)")
             analytics_output = gr.JSON(label="Performance Analytics")
             analytics_btn.click(
-                fn=lambda days: client.get_student_analytics(student_id, days),
+                fn=lambda x: asyncio.run(client.get_student_analytics("student_12345", x)),
                 inputs=[timeframe],
                 outputs=[analytics_output]
             )
@@ -188,33 +184,10 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
             error_output = gr.JSON(label="Error Pattern Analysis")
             
             error_btn.click(
-                fn=lambda concept: client.analyze_error_patterns(student_id, concept),
+                fn=lambda x: asyncio.run(client.analyze_error_patterns("student_12345", x)),
                 inputs=[error_concept],
                 outputs=[error_output]
             )
-        
-        # Tab 5: Assessment Tools
-        with gr.Tab("Assessment Tools"):
-            gr.Markdown("## Create Assessment")
-            
-            with gr.Row():
-                with gr.Column():
-                    assess_concepts = gr.CheckboxGroup(
-                        choices=["math_algebra_basics", "math_algebra_linear_equations", "math_algebra_quadratic_equations"],
-                        label="Select Concepts",
-                        value=["math_algebra_linear_equations"]
-                    )
-                    assess_questions = gr.Slider(minimum=1, maximum=10, value=3, step=1, label="Number of Questions")
-                    assess_diff = gr.Slider(minimum=1, maximum=5, value=3, step=1, label="Difficulty")
-                    create_assess_btn = gr.Button("Create Assessment")
-                
-                with gr.Column():
-                    assessment_output = gr.JSON(label="Generated Assessment")
-            create_assess_btn.click(
-                fn=lambda concepts, num, diff: client.create_assessment(concepts, num, diff),
-                inputs=[assess_concepts, assess_questions, assess_diff],
-                outputs=[assessment_output]
-            )
             
             gr.Markdown("## Plagiarism Detection")
             
@@ -236,11 +209,11 @@ with gr.Blocks(title="TutorX Educational AI", theme=gr.themes.Soft()) as demo:
                     plagiarism_output = gr.JSON(label="Originality Report")
             
             plagiarism_btn.click(
-                fn=lambda sub, ref: client.check_submission_originality(sub, [ref]),
+                fn=lambda x, y: asyncio.run(client.check_submission_originality(x, [y])),
                 inputs=[submission_input, reference_input],
                 outputs=[plagiarism_output]
             )
 
-# Launch the app
+# Launch the interface
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch(server_name="0.0.0.0", server_port=7860)

client.py

@@ -5,21 +5,36 @@ for use by the Gradio interface.
 """
 
 import json
-import requests
+import aiohttp
+import asyncio
 from typing import Dict, Any, List, Optional
 import base64
 from datetime import datetime
+import os
 
-# Default MCP server URL
-MCP_SERVER_URL = "http://localhost:8000"
+# 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=MCP_SERVER_URL):
+    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"
+                }
+            )
     
-    def _call_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
+    async def _call_tool(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
         """
         Call an MCP tool on the server
         
@@ -30,21 +45,22 @@ class TutorXClient:
         Returns:
             Tool response
         """
+        await self._ensure_session()
         try:
-            response = requests.post(
-                f"{self.server_url}/tools/{tool_name}",
+            async with self.session.post(
+                f"{self.server_url}/mcp/tools/{tool_name}",
                 json=params,
-                headers={"Content-Type": "application/json"}
-            )
-            response.raise_for_status()
-            return response.json()
-        except requests.RequestException as e:
+                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()
             }
     
-    def _get_resource(self, resource_uri: str) -> Dict[str, Any]:
+    async def _get_resource(self, resource_uri: str) -> Dict[str, Any]:
         """
         Get an MCP resource from the server
         
@@ -54,72 +70,90 @@ class TutorXClient:
         Returns:
             Resource data
         """
+        await self._ensure_session()
         try:
-            response = requests.get(
-                f"{self.server_url}/resources?uri={resource_uri}",
-                headers={"Accept": "application/json"}
-            )
-            response.raise_for_status()
-            return response.json()
-        except requests.RequestException as e:
+            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 ------------
     
-    def assess_skill(self, student_id: str, concept_id: str) -> Dict[str, Any]:
+    async def assess_skill(self, student_id: str, concept_id: str) -> Dict[str, Any]:
         """Assess student's skill level on a specific concept"""
-        return self._call_tool("assess_skill", {
+        return await self._call_tool("assess_skill", {
             "student_id": student_id,
             "concept_id": concept_id
         })
     
-    def get_concept_graph(self) -> Dict[str, Any]:
+    async def get_concept_graph(self) -> Dict[str, Any]:
         """Get the full knowledge concept graph"""
-        return self._get_resource("concept-graph://")
+        return await self._get_resource("concept-graph://")
     
-    def get_learning_path(self, student_id: str) -> Dict[str, Any]:
+    async def get_learning_path(self, student_id: str) -> Dict[str, Any]:
         """Get personalized learning path for a student"""
-        return self._get_resource(f"learning-path://{student_id}")
+        return await self._get_resource(f"learning-path://{student_id}")
     
-    def generate_quiz(self, concept_ids: List[str], difficulty: int = 2) -> Dict[str, Any]:
+    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 self._call_tool("generate_quiz", {
+        return await self._call_tool("generate_quiz", {
             "concept_ids": concept_ids,
             "difficulty": difficulty
         })
     
-    def analyze_error_patterns(self, student_id: str, concept_id: str) -> Dict[str, Any]:
+    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 self._call_tool("analyze_error_patterns", {
+        return await self._call_tool("analyze_error_patterns", {
             "student_id": student_id,
             "concept_id": concept_id
         })
     
     # ------------ Advanced Features ------------
     
-    def analyze_cognitive_state(self, eeg_data: Dict[str, Any]) -> Dict[str, Any]:
+    async def analyze_cognitive_state(self, eeg_data: Dict[str, Any]) -> Dict[str, Any]:
         """Analyze EEG data to determine cognitive state"""
-        return self._call_tool("analyze_cognitive_state", {
+        return await self._call_tool("analyze_cognitive_state", {
             "eeg_data": eeg_data
         })
     
-    def get_curriculum_standards(self, country_code: str) -> Dict[str, Any]:
+    async def get_curriculum_standards(self, country_code: str) -> Dict[str, Any]:
         """Get curriculum standards for a specific country"""
-        return self._get_resource(f"curriculum-standards://{country_code}")
+        return await self._get_resource(f"curriculum-standards://{country_code}")
     
-    def align_content_to_standard(self, content_id: str, standard_id: str) -> Dict[str, Any]:
+    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 self._call_tool("align_content_to_standard", {
+        return await self._call_tool("align_content_to_standard", {
             "content_id": content_id,
             "standard_id": standard_id
         })
     
-    def generate_lesson(self, topic: str, grade_level: int, duration_minutes: int = 45) -> Dict[str, Any]:
+    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 self._call_tool("generate_lesson", {
+        return await self._call_tool("generate_lesson", {
             "topic": topic,
             "grade_level": grade_level,
             "duration_minutes": duration_minutes
@@ -127,76 +161,83 @@ class TutorXClient:
     
     # ------------ User Experience ------------
     
-    def get_student_dashboard(self, student_id: str) -> Dict[str, Any]:
+    async def get_student_dashboard(self, student_id: str) -> Dict[str, Any]:
         """Get dashboard data for a specific student"""
-        return self._get_resource(f"student-dashboard://{student_id}")
+        return await self._get_resource(f"student-dashboard://{student_id}")
     
-    def get_accessibility_settings(self, student_id: str) -> Dict[str, Any]:
+    async def get_accessibility_settings(self, student_id: str) -> Dict[str, Any]:
         """Get accessibility settings for a student"""
-        return self._call_tool("get_accessibility_settings", {
+        return await self._call_tool("get_accessibility_settings", {
             "student_id": student_id
         })
     
-    def update_accessibility_settings(self, student_id: str, settings: Dict[str, Any]) -> Dict[str, Any]:
+    async def update_accessibility_settings(self, student_id: str, settings: Dict[str, Any]) -> Dict[str, Any]:
         """Update accessibility settings for a student"""
-        return self._call_tool("update_accessibility_settings", {
+        return await self._call_tool("update_accessibility_settings", {
             "student_id": student_id,
             "settings": settings
         })
     
     # ------------ Multi-Modal Interaction ------------
     
-    def text_interaction(self, query: str, student_id: str) -> Dict[str, Any]:
+    async def text_interaction(self, query: str, student_id: str) -> Dict[str, Any]:
         """Process a text query from the student"""
-        return self._call_tool("text_interaction", {
+        return await self._call_tool("text_interaction", {
             "query": query,
             "student_id": student_id
         })
     
-    def voice_interaction(self, audio_data_base64: str, student_id: str) -> Dict[str, Any]:
+    async def voice_interaction(self, audio_data_base64: str, student_id: str) -> Dict[str, Any]:
         """Process voice input from the student"""
-        return self._call_tool("voice_interaction", {
+        return await self._call_tool("voice_interaction", {
             "audio_data_base64": audio_data_base64,
             "student_id": student_id
         })
     
-    def handwriting_recognition(self, image_data_base64: str, student_id: str) -> Dict[str, Any]:
+    async def handwriting_recognition(self, image_data_base64: str, student_id: str) -> Dict[str, Any]:
         """Process handwritten input from the student"""
-        return self._call_tool("handwriting_recognition", {
+        return await self._call_tool("handwriting_recognition", {
             "image_data_base64": image_data_base64,
             "student_id": student_id
         })
     
     # ------------ Assessment ------------
     
-    def create_assessment(self, concept_ids: List[str], num_questions: int, difficulty: int = 3) -> Dict[str, Any]:
+    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 self._call_tool("create_assessment", {
+        return await self._call_tool("create_assessment", {
             "concept_ids": concept_ids,
             "num_questions": num_questions,
             "difficulty": difficulty
         })
     
-    def grade_assessment(self, assessment_id: str, student_answers: Dict[str, str], questions: List[Dict[str, Any]]) -> Dict[str, Any]:
+    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 self._call_tool("grade_assessment", {
+        return await self._call_tool("grade_assessment", {
             "assessment_id": assessment_id,
             "student_answers": student_answers,
             "questions": questions
         })
     
-    def get_student_analytics(self, student_id: str, timeframe_days: int = 30) -> Dict[str, Any]:
+    async def get_student_analytics(self, student_id: str, timeframe_days: int = 30) -> Dict[str, Any]:
         """Get comprehensive analytics for a student"""
-        return self._call_tool("get_student_analytics", {
+        return await self._call_tool("get_student_analytics", {
             "student_id": student_id,
             "timeframe_days": timeframe_days
-        })    
-    def check_submission_originality(self, submission: str, reference_sources: List[str]) -> Dict[str, Any]:
+        })
+    
+    async def check_submission_originality(self, submission: str, reference_sources: List[str]) -> Dict[str, Any]:
         """Check student submission for potential plagiarism"""
-        return self._call_tool("check_submission_originality", {
+        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()

docs/API.md

@@ -0,0 +1,178 @@
+# TutorX MCP API Documentation
+
+## Overview
+
+The TutorX MCP API provides a comprehensive set of endpoints for educational tools and resources. The API follows RESTful principles and uses JSON for request/response bodies.
+
+## Base URL
+
+```
+http://127.0.0.1:8000
+```
+
+## Authentication
+
+Currently, the API does not require authentication. This will be implemented in future versions.
+
+## Endpoints
+
+### Health Check
+
+```http
+GET /health
+```
+
+Returns the health status of the server.
+
+**Response**
+```json
+{
+    "status": "healthy",
+    "timestamp": "2024-03-14T12:00:00.000Z",
+    "server": "TutorX MCP",
+    "version": "1.0.0"
+}
+```
+
+### Core Features
+
+#### Assess Skill
+
+```http
+POST /mcp/tools/assess_skill
+```
+
+Assesses a student's skill level on a specific concept.
+
+**Request Body**
+```json
+{
+    "student_id": "string",
+    "concept_id": "string"
+}
+```
+
+**Response**
+```json
+{
+    "student_id": "string",
+    "concept_id": "string",
+    "skill_level": 0.75,
+    "confidence": 0.85,
+    "recommendations": [
+        "Practice more complex problems",
+        "Review related concept: algebra_linear_equations"
+    ],
+    "timestamp": "2024-03-14T12:00:00.000Z"
+}
+```
+
+#### Get Concept Graph
+
+```http
+GET /mcp/resources/concept-graph://
+```
+
+Returns the full knowledge concept graph.
+
+**Response**
+```json
+{
+    "nodes": [
+        {
+            "id": "math_algebra_basics",
+            "name": "Algebra Basics",
+            "difficulty": 1
+        }
+    ],
+    "edges": [
+        {
+            "from": "math_algebra_basics",
+            "to": "math_algebra_linear_equations",
+            "weight": 1.0
+        }
+    ]
+}
+```
+
+#### Get Learning Path
+
+```http
+GET /mcp/resources/learning-path://{student_id}
+```
+
+Returns a personalized learning path for a student.
+
+**Response**
+```json
+{
+    "student_id": "string",
+    "current_concepts": ["math_algebra_linear_equations"],
+    "recommended_next": ["math_algebra_quadratic_equations"],
+    "mastered": ["math_algebra_basics"],
+    "estimated_completion_time": "2 weeks"
+}
+```
+
+#### Generate Quiz
+
+```http
+POST /mcp/tools/generate_quiz
+```
+
+Generates a quiz based on specified concepts and difficulty.
+
+**Request Body**
+```json
+{
+    "concept_ids": ["string"],
+    "difficulty": 2
+}
+```
+
+**Response**
+```json
+{
+    "quiz_id": "string",
+    "concept_ids": ["string"],
+    "difficulty": 2,
+    "questions": [
+        {
+            "id": "string",
+            "text": "string",
+            "type": "string",
+            "answer": "string",
+            "solution_steps": ["string"]
+        }
+    ]
+}
+```
+
+## Error Responses
+
+The API uses standard HTTP status codes and returns error details in the response body:
+
+```json
+{
+    "error": "Error message",
+    "timestamp": "2024-03-14T12:00:00.000Z"
+}
+```
+
+Common status codes:
+- 200: Success
+- 400: Bad Request
+- 404: Not Found
+- 500: Internal Server Error
+
+## Rate Limiting
+
+Currently, there are no rate limits implemented. This will be added in future versions.
+
+## Versioning
+
+The API version is included in the response headers and health check endpoint. Future versions will support versioning through the URL path.
+
+## Support
+
+For support or to report issues, please contact the development team or create an issue in the project repository. 

main.py

@@ -2,8 +2,14 @@
 from mcp.server.fastmcp import FastMCP
 import json
 import os
+import warnings
 from typing import List, Dict, Any, Optional
 from datetime import datetime
+from fastapi import FastAPI, Query
+from fastapi.responses import JSONResponse
+
+# Filter out the tool registration warning
+warnings.filterwarnings("ignore", message="Tool already exists")
 
 # Import utility functions
 from utils.multimodal import (
@@ -19,14 +25,29 @@ from utils.assessment import (
     detect_plagiarism
 )
 
-# Create the TutorX MCP server
-mcp = FastMCP("TutorX")
+# Get server configuration from environment variables with defaults
+SERVER_HOST = os.getenv("MCP_HOST", "0.0.0.0")  # Allow connections from any IP
+SERVER_PORT = int(os.getenv("MCP_PORT", "8001"))
+SERVER_TRANSPORT = os.getenv("MCP_TRANSPORT", "http")
+
+# Create the TutorX MCP server with explicit configuration
+mcp = FastMCP(
+    "TutorX",
+    dependencies=["mcp[cli]>=1.9.3", "gradio>=4.19.0", "numpy>=1.24.0", "pillow>=10.0.0"],
+    host=SERVER_HOST,
+    port=SERVER_PORT,
+    transport=SERVER_TRANSPORT,
+    cors_origins=["*"]  # Allow CORS from any origin
+)
+
+# Create FastAPI app
+api_app = FastAPI()
 
 # ------------------ Core Features ------------------
 
 # Adaptive Learning Engine
 @mcp.tool()
-def assess_skill(student_id: str, concept_id: str) -> Dict[str, Any]:
+async def assess_skill(student_id: str, concept_id: str) -> Dict[str, Any]:
     """
     Assess student's skill level on a specific concept
     
@@ -37,21 +58,27 @@ def assess_skill(student_id: str, concept_id: str) -> Dict[str, Any]:
     Returns:
         Dictionary containing skill level and recommendations
     """
-    # Simulated skill assessment
-    return {
-        "student_id": student_id,
-        "concept_id": concept_id,
-        "skill_level": 0.75,
-        "confidence": 0.85,
-        "recommendations": [
-            "Practice more complex problems",
-            "Review related concept: algebra_linear_equations"
-        ],
-        "timestamp": datetime.now().isoformat()
-    }
+    try:
+        # Simulated skill assessment
+        return {
+            "student_id": student_id,
+            "concept_id": concept_id,
+            "skill_level": 0.75,
+            "confidence": 0.85,
+            "recommendations": [
+                "Practice more complex problems",
+                "Review related concept: algebra_linear_equations"
+            ],
+            "timestamp": datetime.now().isoformat()
+        }
+    except Exception as e:
+        return {
+            "error": str(e),
+            "timestamp": datetime.now().isoformat()
+        }
 
 @mcp.resource("concept-graph://")
-def get_concept_graph() -> Dict[str, Any]:
+async def get_concept_graph() -> Dict[str, Any]:
     """Get the full knowledge concept graph"""
     return {
         "nodes": [
@@ -66,7 +93,7 @@ def get_concept_graph() -> Dict[str, Any]:
     }
 
 @mcp.resource("learning-path://{student_id}")
-def get_learning_path(student_id: str) -> Dict[str, Any]:
+async def get_learning_path(student_id: str) -> Dict[str, Any]:
     """Get personalized learning path for a student"""
     return {
         "student_id": student_id,
@@ -78,7 +105,7 @@ def get_learning_path(student_id: str) -> Dict[str, Any]:
 
 # Assessment Suite
 @mcp.tool()
-def generate_quiz(concept_ids: List[str], difficulty: int = 2) -> Dict[str, Any]:
+async def generate_quiz(concept_ids: List[str], difficulty: int = 2) -> Dict[str, Any]:
     """
     Generate a quiz based on specified concepts and difficulty
     
@@ -109,476 +136,28 @@ def generate_quiz(concept_ids: List[str], difficulty: int = 2) -> Dict[str, Any]
         ]
     }
 
-# Feedback System
-@mcp.tool()
-def analyze_error_patterns(student_id: str, concept_id: str) -> Dict[str, Any]:
-    """
-    Analyze common error patterns for a student on a specific concept
-    
-    Args:
-        student_id: The student's unique identifier
-        concept_id: The concept to analyze
-        
-    Returns:
-        Error pattern analysis
-    """
-    return {
-        "student_id": student_id,
-        "concept_id": concept_id,
-        "common_errors": [
-            {
-                "type": "sign_error",
-                "frequency": 0.65,
-                "example": "2x - 3 = 5 → 2x = 5 - 3 → 2x = 2 → x = 1 (should be x = 4)"
-            },
-            {
-                "type": "arithmetic_error",
-                "frequency": 0.35,
-                "example": "2x = 8 → x = 8/2 = 3 (should be x = 4)"
-            }
-        ],
-        "recommendations": [
-            "Practice more sign manipulation problems",
-            "Review basic arithmetic operations"
-        ]
-    }
-
-# ------------------ Advanced Features ------------------
-
-# Neurological Engagement Monitor
-@mcp.tool()
-def analyze_cognitive_state(eeg_data: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Analyze EEG data to determine cognitive state
-    
-    Args:
-        eeg_data: Raw or processed EEG data
-        
-    Returns:
-        Analysis of cognitive state
-    """
-    return {
-        "attention_level": 0.82,
-        "cognitive_load": 0.65,
-        "stress_level": 0.25,
-        "recommendations": [
-            "Student is engaged but approaching cognitive overload",
-            "Consider simplifying next problems slightly"
-        ],
-        "timestamp": datetime.now().isoformat()
-    }
-
-# Cross-Institutional Knowledge Fusion
-@mcp.resource("curriculum-standards://{country_code}")
-def get_curriculum_standards(country_code: str) -> Dict[str, Any]:
-    """Get curriculum standards for a specific country"""
-    standards = {
-        "us": {
-            "name": "Common Core State Standards",
-            "math_standards": {
-                "algebra_1": [
-                    "CCSS.Math.Content.HSA.CED.A.1",
-                    "CCSS.Math.Content.HSA.CED.A.2"
-                ]
-            }
-        },
-        "uk": {
-            "name": "National Curriculum",
-            "math_standards": {
-                "algebra_1": [
-                    "KS3.Algebra.1",
-                    "KS3.Algebra.2"
-                ]
-            }
-        }
-    }
-    
-    return standards.get(country_code.lower(), {"error": "Country code not found"})
-
-@mcp.tool()
-def align_content_to_standard(content_id: str, standard_id: str) -> Dict[str, Any]:
-    """
-    Align educational content to a specific curriculum standard
-    
-    Args:
-        content_id: The ID of the content to align
-        standard_id: The curriculum standard ID
-        
-    Returns:
-        Alignment information and recommendations
-    """
-    return {
-        "content_id": content_id,
-        "standard_id": standard_id,
-        "alignment_score": 0.85,
-        "gaps": [
-            "Missing coverage of polynomial division",
-            "Should include more word problems"
-        ],
-        "recommendations": [
-            "Add 2-3 examples of polynomial division",
-            "Convert 30% of problems to word problems"
-        ]
-    }
-
-# Automated Lesson Authoring
-@mcp.tool()
-def generate_lesson(topic: str, grade_level: int, duration_minutes: int = 45) -> Dict[str, Any]:
-    """
-    Generate a complete lesson plan on a topic
-    
-    Args:
-        topic: The lesson topic
-        grade_level: Target grade level (K-12)
-        duration_minutes: Lesson duration in minutes
-        
-    Returns:
-        Complete lesson plan
-    """
-    return {
-        "topic": topic,
-        "grade_level": grade_level,
-        "duration_minutes": duration_minutes,
-        "objectives": [
-            "Students will be able to solve linear equations in one variable",
-            "Students will be able to check their solutions"
-        ],
-        "materials": [
-            "Whiteboard/projector",
-            "Handouts with practice problems",
-            "Graphing calculators (optional)"
-        ],
-        "activities": [
-            {
-                "name": "Warm-up",
-                "duration_minutes": 5,
-                "description": "Review of pre-algebra concepts needed for today's lesson"
-            },
-            {
-                "name": "Direct Instruction",
-                "duration_minutes": 15,
-                "description": "Teacher demonstrates solving linear equations step by step"
-            },
-            {
-                "name": "Guided Practice",
-                "duration_minutes": 10,
-                "description": "Students solve problems with teacher guidance"
-            },
-            {
-                "name": "Independent Practice",
-                "duration_minutes": 10,
-                "description": "Students solve problems independently"
-            },
-            {
-                "name": "Closure",
-                "duration_minutes": 5,
-                "description": "Review key concepts and preview next lesson"
-            }
-        ],
-        "assessment": {
-            "formative": "Teacher observation during guided and independent practice",
-            "summative": "Exit ticket with 3 problems to solve"
-        },
-        "differentiation": {
-            "struggling": "Provide equation-solving steps reference sheet",
-            "advanced": "Offer multi-step equations with fractions and decimals"
-        }
-    }
-
-# ------------------ User Experience Features ------------------
-
-@mcp.resource("student-dashboard://{student_id}")
-def get_student_dashboard(student_id: str) -> Dict[str, Any]:
-    """Get dashboard data for a specific student"""
-    return {
-        "student_id": student_id,
-        "knowledge_map": {
-            "mastery_percentage": 68,
-            "concepts_mastered": 42,
-            "concepts_in_progress": 15,
-            "concepts_not_started": 25
-        },
-        "recent_activity": [
-            {
-                "timestamp": "2025-06-06T15:30:00Z",
-                "activity_type": "quiz",
-                "description": "Algebra Quiz #3",
-                "performance": "85%"
-            },
-            {
-                "timestamp": "2025-06-05T13:45:00Z",
-                "activity_type": "lesson",
-                "description": "Quadratic Equations Introduction",
-                "duration_minutes": 32
-            }
-        ],
-        "recommendations": [
-            "Complete Factor Polynomials practice set",
-            "Review Linear Systems interactive module"
-        ]
-    }
-
-@mcp.tool()
-def get_accessibility_settings(student_id: str) -> Dict[str, Any]:
-    """
-    Get accessibility settings for a student
-    
-    Args:
-        student_id: The student's unique identifier
-        
-    Returns:
-        Accessibility settings
-    """
-    return {
-        "student_id": student_id,
-        "text_to_speech_enabled": True,
-        "font_size": "large",
-        "high_contrast_mode": False,
-        "screen_reader_compatible": True,
-        "keyboard_navigation_enabled": True
-    }
-
-@mcp.tool()
-def update_accessibility_settings(student_id: str, settings: Dict[str, Any]) -> Dict[str, Any]:
-    """
-    Update accessibility settings for a student
-    
-    Args:
-        student_id: The student's unique identifier
-        settings: Dictionary of settings to update
-        
-    Returns:
-        Updated accessibility settings
-    """
-    # In a real implementation, this would update a database
-    return {
-        "student_id": student_id,
-        "text_to_speech_enabled": settings.get("text_to_speech_enabled", True),
-        "font_size": settings.get("font_size", "large"),
-        "high_contrast_mode": settings.get("high_contrast_mode", False),
-        "screen_reader_compatible": settings.get("screen_reader_compatible", True),
-        "keyboard_navigation_enabled": settings.get("keyboard_navigation_enabled", True),
-        "updated_at": datetime.now().isoformat()
-    }
-
-# ------------------ Multi-Modal Interaction ------------------
-
-@mcp.tool()
-@mcp.tool()
-def text_interaction(query: str, student_id: str, session_context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
-    """
-    Process a text query from the student
-    
-    Args:
-        query: The text query from the student
-        student_id: The student's unique identifier
-        session_context: Optional context about the current session
-        
-    Returns:
-        Processed response
-    """
-    # Add student information to context
-    context = session_context or {}
-    context["student_id"] = student_id
-    
-    return process_text_query(query, context)
-
-@mcp.tool()
-def voice_interaction(audio_data_base64: str, student_id: str) -> Dict[str, Any]:
-    """
-    Process voice input from the student
-    
-    Args:
-        audio_data_base64: Base64 encoded audio data
-        student_id: The student's unique identifier
-        
-    Returns:
-        Transcription and response
-    """
-    # Process voice input
-    result = process_voice_input(audio_data_base64)
-    
-    # Process the transcription as a text query
-    text_response = process_text_query(result["transcription"], {"student_id": student_id})
-    
-    # Generate speech response
-    speech_response = generate_speech_response(
-        text_response["response"],
-        {"voice_id": "educational_tutor"}
-    )
-    
-    # Combine results
-    return {
-        "input_transcription": result["transcription"],
-        "input_confidence": result["confidence"],
-        "detected_emotions": result.get("detected_emotions", {}),
-        "text_response": text_response["response"],
-        "speech_response": speech_response,
-        "timestamp": datetime.now().isoformat()
-    }
-
-@mcp.tool()
-def handwriting_recognition(image_data_base64: str, student_id: str) -> Dict[str, Any]:
-    """
-    Process handwritten input from the student
-    
-    Args:
-        image_data_base64: Base64 encoded image data of handwriting
-        student_id: The student's unique identifier
-        
-    Returns:
-        Transcription and analysis
-    """
-    # Process handwriting input
-    result = process_handwriting(image_data_base64)
-    
-    # If it's a math equation, solve it
-    if result["detected_content_type"] == "math_equation":
-        # In a real implementation, this would use a math engine to solve the equation
-        # For demonstration, we'll provide a simulated solution
-        if result["equation_type"] == "quadratic":
-            solution = {
-                "equation": result["transcription"],
-                "solution_steps": [
-                    "x^2 + 5x + 6 = 0",
-                    "Factor: (x + 2)(x + 3) = 0",
-                    "x + 2 = 0 or x + 3 = 0",
-                    "x = -2 or x = -3"
-                ],
-                "solutions": [-2, -3]
-            }
-        else:
-            solution = {
-                "equation": result["transcription"],
-                "note": "Solution not implemented for this equation type"
-            }
-    else:
-        solution = None
-    
-    return {
-        "transcription": result["transcription"],
-        "confidence": result["confidence"],
-        "detected_content_type": result["detected_content_type"],
-        "solution": solution,
-        "timestamp": datetime.now().isoformat()
-    }
-
-# ------------------ Advanced Assessment Tools ------------------
-
-@mcp.tool()
-def create_assessment(concept_ids: List[str], num_questions: int, difficulty: int = 3) -> Dict[str, Any]:
-    """
-    Create a complete assessment for given concepts
-    
-    Args:
-        concept_ids: List of concept IDs to include
-        num_questions: Number of questions to generate
-        difficulty: Difficulty level (1-5)
-        
-    Returns:
-        Complete assessment with questions
-    """
-    questions = []
-    
-    # Distribute questions evenly among concepts
-    questions_per_concept = num_questions // len(concept_ids)
-    extra_questions = num_questions % len(concept_ids)
-    
-    for i, concept_id in enumerate(concept_ids):
-        # Determine how many questions for this concept
-        concept_questions = questions_per_concept
-        if i < extra_questions:
-            concept_questions += 1
-        
-        # Generate questions for this concept
-        for _ in range(concept_questions):
-            questions.append(generate_question(concept_id, difficulty))
-    
-    return {
-        "assessment_id": f"assessment_{datetime.now().strftime('%Y%m%d%H%M%S')}",
-        "concept_ids": concept_ids,
-        "difficulty": difficulty,
-        "num_questions": len(questions),
-        "questions": questions,
-        "created_at": datetime.now().isoformat()
-    }
+@api_app.get("/api/assess_skill")
+async def assess_skill_api(student_id: str = Query(...), concept_id: str = Query(...)):
+    result = await assess_skill(student_id, concept_id)
+    return JSONResponse(content=result)
 
-@mcp.tool()
-def grade_assessment(assessment_id: str, student_answers: Dict[str, str], questions: List[Dict[str, Any]]) -> Dict[str, Any]:
-    """
-    Grade a completed assessment
-    
-    Args:
-        assessment_id: The ID of the assessment
-        student_answers: Dictionary mapping question IDs to student answers
-        questions: List of question objects
-        
-    Returns:
-        Grading results
-    """
-    results = []
-    correct_count = 0
-    
-    for question in questions:
-        question_id = question["id"]
-        if question_id in student_answers:
-            evaluation = evaluate_student_answer(question, student_answers[question_id])
-            results.append(evaluation)
-            if evaluation["is_correct"]:
-                correct_count += 1
-    
-    # Calculate score
-    score = correct_count / len(questions) if questions else 0
-    
-    # Analyze error patterns
-    error_types = {}
-    for result in results:
-        if result["error_type"]:
-            error_type = result["error_type"]
-            error_types[error_type] = error_types.get(error_type, 0) + 1
-    
-    # Find most common error
-    most_common_error = None
-    if error_types:
-        most_common_error = max(error_types.items(), key=lambda x: x[1])
-    
-    return {
-        "assessment_id": assessment_id,
-        "score": score,
-        "correct_count": correct_count,
-        "total_questions": len(questions),
-        "results": results,
-        "most_common_error": most_common_error,
-        "completed_at": datetime.now().isoformat()
-    }
+@api_app.post("/api/generate_quiz")
+async def generate_quiz_api(concept_ids: list[str], difficulty: int = 2):
+    result = await generate_quiz(concept_ids, difficulty)
+    return JSONResponse(content=result)
 
-@mcp.tool()
-def get_student_analytics(student_id: str, timeframe_days: int = 30) -> Dict[str, Any]:
-    """
-    Get comprehensive analytics for a student
-    
-    Args:
-        student_id: The student's unique identifier
-        timeframe_days: Number of days to include in analysis
-        
-    Returns:
-        Performance analytics
-    """
-    return generate_performance_analytics(student_id, timeframe_days)
+# Mount FastAPI app to MCP server
+mcp.app = api_app
 
-@mcp.tool()
-def check_submission_originality(submission: str, reference_sources: List[str]) -> Dict[str, Any]:
-    """
-    Check student submission for potential plagiarism
-    
-    Args:
-        submission: The student's submission text
-        reference_sources: List of reference texts to check against
-        
-    Returns:
-        Originality analysis
-    """
-    return detect_plagiarism(submission, reference_sources)
+# Function to run the server
+def run_server():
+    """Run the MCP server with configured transport and port"""
+    print(f"Starting TutorX MCP Server on {SERVER_HOST}:{SERVER_PORT} using {SERVER_TRANSPORT} transport...")
+    try:
+        mcp.run(transport="sse")
+    except Exception as e:
+        print(f"Error starting server: {str(e)}")
+        raise
 
 if __name__ == "__main__":
-    mcp.run()
+    run_server()

pyproject.toml

@@ -6,18 +6,25 @@ readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
     "mcp[cli]>=1.9.3",
+    "fastapi>=0.109.0",
+    "uvicorn>=0.27.0",
+    "aiohttp>=3.9.0",
     "gradio>=4.19.0",
     "numpy>=1.24.0",
     "pillow>=10.0.0",
-    "requests>=2.31.0",
+    "python-multipart>=0.0.6",
+    "pydantic>=2.6.0",
 ]
 
 [project.optional-dependencies]
 test = [
     "pytest>=7.4.0",
     "pytest-cov>=4.1.0",
+    "pytest-asyncio>=0.23.0",
+    "httpx>=0.26.0",
 ]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 python_files = "test_*.py"
+asyncio_mode = "auto"

requirements.txt

@@ -0,0 +1,19 @@
+fastapi>=0.68.0
+uvicorn>=0.15.0
+aiohttp>=3.8.0
+python-multipart>=0.0.5
+pydantic>=1.8.0
+gradio>=4.19.0
+numpy>=1.24.0
+pillow>=10.0.0
+python-jose[cryptography]>=3.3.0
+passlib[bcrypt]>=1.7.4
+python-dotenv>=0.19.0
+httpx>=0.24.0
+pytest>=7.0.0
+pytest-asyncio>=0.18.0
+pytest-cov>=3.0.0
+black>=22.0.0
+isort>=5.10.0
+mypy>=0.910
+ruff>=0.0.262 

run.py

@@ -6,6 +6,8 @@ import argparse
 import importlib.util
 import os
 import sys
+import time
+import subprocess
 
 def load_module(name, path):
     """Load a module from path"""
@@ -14,16 +16,22 @@ def load_module(name, path):
     spec.loader.exec_module(module)
     return module
 
-def run_mcp_server():
-    """Run the MCP server"""
-    print("Starting TutorX MCP Server...")
+def run_mcp_server(host="127.0.0.1", port=8000, transport="streamable-http"):
+    """Run the MCP server with specified configuration"""
+    print(f"Starting TutorX MCP Server on {host}:{port} using {transport} transport...")
+    
+    # Set environment variables for MCP server
+    os.environ["MCP_HOST"] = host
+    os.environ["MCP_PORT"] = str(port)
+    os.environ["MCP_TRANSPORT"] = transport
+    
     main_module = load_module("main", "main.py")
     
     # Access the mcp instance and run it
-    if hasattr(main_module, "mcp"):
-        main_module.mcp.run()
+    if hasattr(main_module, "run_server"):
+        main_module.run_server()
     else:
-        print("Error: MCP server instance not found in main.py")
+        print("Error: run_server function not found in main.py")
         sys.exit(1)
 
 def run_gradio_interface():
@@ -38,6 +46,17 @@ def run_gradio_interface():
         print("Error: Gradio demo not found in app.py")
         sys.exit(1)
 
+def check_port_available(port):
+    """Check if a port is available"""
+    import socket
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    try:
+        sock.bind(('127.0.0.1', port))
+        sock.close()
+        return True
+    except:
+        return False
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Run TutorX MCP Server or Gradio Interface")
     parser.add_argument(
@@ -57,24 +76,36 @@ if __name__ == "__main__":
         default=8000,
         help="Port to use"
     )
+    parser.add_argument(
+        "--transport",
+        choices=["stdio", "streamable-http", "sse"],
+        default="streamable-http",
+        help="Transport protocol to use"
+    )
     
     args = parser.parse_args()
     
     if args.mode == "mcp":
-        # Set environment variables for MCP server
-        os.environ["MCP_HOST"] = args.host
-        os.environ["MCP_PORT"] = str(args.port)
-        run_mcp_server()
+        if not check_port_available(args.port):
+            print(f"Warning: Port {args.port} is already in use. Trying to use the server anyway...")
+        run_mcp_server(args.host, args.port, args.transport)
     elif args.mode == "gradio":
         run_gradio_interface()
     elif args.mode == "both":
         # For 'both' mode, we'll start MCP server in a separate process
-        import subprocess
-        import time
-        
+        if not check_port_available(args.port):
+            print(f"Warning: Port {args.port} is already in use. Trying to use the server anyway...")
+            
         # Start MCP server in a background process
         mcp_process = subprocess.Popen(
-            [sys.executable, "run.py", "--mode", "mcp", "--host", args.host, "--port", str(args.port)],
+            [
+                sys.executable, 
+                "run.py", 
+                "--mode", "mcp", 
+                "--host", args.host, 
+                "--port", str(args.port),
+                "--transport", args.transport
+            ],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE
         )