Spaces:
Sleeping
Sleeping
File size: 7,421 Bytes
1af10cc a806ca2 1af10cc a806ca2 1af10cc a806ca2 1af10cc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 |
"""
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")
|