""" TutorX MCP Server This is the main entry point for the TutorX MCP server. """ import base64 import os import sys from pathlib import Path # Add the current directory to the Python path current_dir = Path(__file__).parent sys.path.insert(0, str(current_dir)) import uvicorn from fastapi import FastAPI, HTTPException, UploadFile, File, Form from fastapi.middleware.cors import CORSMiddleware from mcp.server.fastmcp import FastMCP # Import all tools to register them with MCP from tools import ( concept_tools, lesson_tools, quiz_tools, interaction_tools, ocr_tools, learning_path_tools ) # Import resources from resources import concept_graph, curriculum_standards # Create FastAPI app api_app = FastAPI( title="TutorX MCP Server", description="Model Context Protocol server for TutorX educational platform", version="1.0.0" ) # Add CORS middleware api_app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Import the shared mcp instance from mcp_server.mcp_instance import mcp # Explicitly import all tool modules so their @mcp.tool() decorators run from mcp_server.tools import concept_tools from mcp_server.tools import lesson_tools from mcp_server.tools import quiz_tools from mcp_server.tools import interaction_tools from mcp_server.tools import ocr_tools from mcp_server.tools import learning_path_tools from mcp_server.tools import concept_graph_tools # Mount the SSE transport for MCP at '/sse/' (with trailing slash) api_app.mount("/sse", mcp.sse_app()) # Health check endpoint @api_app.get("/health") async def health_check(): return {"status": "healthy", "service": "tutorx-mcp"} # API endpoints - Concepts @api_app.get("/api/concept_graph") async def get_concept_graph(concept_id: str = None): if concept_id: concept = concept_graph.get_concept(concept_id) if not concept: raise HTTPException(status_code=404, detail={"error": f"Concept {concept_id} not found"}) return concept return {"concepts": list(concept_graph.get_concept_graph().values())} @api_app.get("/api/concept/{concept_id}") async def get_concept_endpoint(concept_id: str): concept = concept_graph.get_concept(concept_id) if not concept: raise HTTPException(status_code=404, detail=f"Concept {concept_id} not found") return concept @api_app.get("/api/concepts") async def list_concepts(): return concept_graph.get_all_concepts() # API endpoints - Curriculum Standards @api_app.get("/api/curriculum-standards") async def get_curriculum_standards(country: str = "us"): return curriculum_standards.get_curriculum_standards(country) # API endpoints - Text Interaction @api_app.post("/api/text-interaction") async def text_interaction_endpoint(request: dict): query = request.get("query") student_id = request.get("student_id") if not query or not student_id: raise HTTPException(status_code=400, detail="Both query and student_id are required") return await interaction_tools.text_interaction(query, student_id) # API endpoints - Submission Originality Check @api_app.post("/api/check-originality") async def check_originality_endpoint(request: dict): submission = request.get("submission") reference_sources = request.get("reference_sources", []) if not submission or not isinstance(reference_sources, list): raise HTTPException(status_code=400, detail="submission (string) and reference_sources (array) are required") return await interaction_tools.check_submission_originality(submission, reference_sources) # API endpoints - Document OCR @api_app.post("/api/document-ocr") async def document_ocr_endpoint( file: UploadFile = File(...) ): try: # Save the uploaded file to a temporary location import tempfile import os # Get the file extension file_extension = os.path.splitext(file.filename)[1].lower() # Create a temporary file with the same extension with tempfile.NamedTemporaryFile(delete=False, suffix=file_extension) as temp_file: content = await file.read() temp_file.write(content) temp_file_path = temp_file.name try: # Upload the file to storage and get the URL from mcp_server.utils.azure_upload import upload_to_azure document_url = upload_to_azure(temp_file_path) # Process the document with OCR result = await ocr_tools.mistral_document_ocr(document_url) return result finally: # Clean up the temporary file try: os.unlink(temp_file_path) except: pass except Exception as e: raise HTTPException(status_code=500, detail=f"Error processing document: {str(e)}") # API endpoints - Learning Path @api_app.post("/api/learning-path") async def learning_path_endpoint(request: dict): student_id = request.get("student_id") concept_ids = request.get("concept_ids", []) student_level = request.get("student_level") if not student_id or not concept_ids: raise HTTPException(status_code=400, detail="student_id and concept_ids are required") return await learning_path_tools.get_learning_path( student_id=student_id, concept_ids=concept_ids, student_level=student_level ) # API endpoints - Assess Skill from tools.concept_tools import assess_skill_tool @api_app.post("/api/assess-skill") async def assess_skill_endpoint(request: dict): student_id = request.get("student_id") concept_id = request.get("concept_id") if not student_id or not concept_id: raise HTTPException(status_code=400, detail="Both student_id and concept_id are required") return await assess_skill_tool(student_id, concept_id) # API endpoints - Generate Lesson from tools.lesson_tools import generate_lesson_tool @api_app.post("/api/generate-lesson") async def generate_lesson_endpoint(request: dict): topic = request.get("topic") grade_level = request.get("grade_level") duration_minutes = request.get("duration_minutes") if not topic or grade_level is None or duration_minutes is None: raise HTTPException(status_code=400, detail="topic, grade_level, and duration_minutes are required") return await generate_lesson_tool(topic, grade_level, duration_minutes) # API endpoints - Generate Quiz from tools.quiz_tools import generate_quiz_tool @api_app.post("/api/generate-quiz") async def generate_quiz_endpoint(request: dict): concept = request.get("concept", "") difficulty = request.get("difficulty", 2) if not concept or not isinstance(concept, str) or not concept.strip(): raise HTTPException(status_code=400, detail="concept must be a non-empty string") if isinstance(difficulty, (int, float)): if difficulty <= 2: difficulty = "easy" elif difficulty <= 4: difficulty = "medium" else: difficulty = "hard" if difficulty not in ["easy", "medium", "hard"]: difficulty = "medium" return await generate_quiz_tool(concept.strip(), difficulty) # Entrypoint for running with MCP SSE transport if __name__ == "__main__": mcp.run(transport="sse")