Upload escalation_server.py

escalation_server.py

@@ -0,0 +1,260 @@
+# from mcp.server.fastmcp import FastMCP 
+from fastmcp import FastMCP, Context
+from fastmcp.dependencies import CurrentContext
+import os
+import smtplib
+from email.message import EmailMessage
+import aiosmtplib
+# qqvl obms fhwx llic 
+
+# --- EMAIL CONFIGURATION ---
+SMTP_SERVER = "smtp.gmail.com"
+SMTP_PORT = 587
+SENDER_EMAIL = "dummyuseai@gmail.com"
+SENDER_PASSWORD = "qqvl obms fhwx llic"
+
+# Initialize Server
+mcp = FastMCP("escalation_server")
+
+# Path Configuration
+RESOURCE_PATH = r"E:\Pratham\2026\MCP\resource\jemh114-min (1).md"
+
+# --- CENTRALIZED REGISTRY ---
+FACULTY_DATA = {
+    "Mr. Murty":  {"email": "dharmin.naik@webashlar.com", "id": "T-MURTY-01"},
+    "Mrs. Sunita":{"email": "dummyaiapi03@gmail.com", "id": "T-SUNITA-03"},
+    "Mr. Patel":  {"email": "dev.patel@webashlar.com", "id": "T-PATEL-05"},
+    "Dr. Rao":    {"email": "pratham.lakdawala@webashlar.com", "id": "T-RAO-02"},
+    "Mr. Khanna": {"email": "dummyuseai@gmail.com", "id": "T-KHANNA-04"},
+    "Ms. Joshi":  {"email": "dummyaiapi532@gmail.com", "id": "T-JOSHI-06"}
+}
+
+CLASS_ASSIGNMENTS = {
+    "10-A": {"Math": "Mr. Murty", "Science": "Dr. Rao"},
+    "10-B": {"Math": "Mrs. Sunita", "Science": "Mr. Khanna"},
+    "10-C": {"Math": "Mr. Patel", "Science": "Ms. Joshi"}
+}
+
+# Input Normalization Map
+SUBJECT_MAP = {
+    "mathematics": "Math",
+    "math": "Math",
+    "maths": "Math",
+    "statistics": "Math", 
+    "stats": "Math",      
+    "science": "Science",
+    "physics": "Science", # For this demo, mapping physics to general science
+    "biology": "Science"
+}
+
+def normalize_subject(subject: str) -> str:
+    """Standardizes subject names so 'Mathematics' becomes 'Math'."""
+    return SUBJECT_MAP.get(subject.lower(), subject.title())
+
+# --- 1. MCP RESOURCES ---
+# Resources are "read-only" data sources the LLM can reference.
+def read_resource_file() -> str:
+    if not os.path.exists(RESOURCE_PATH):
+        return "Error: Statistics markdown file not found."
+    with open(RESOURCE_PATH, "r", encoding="utf-8") as f:
+        return f.read()
+
+@mcp.resource("study://math/statistics")
+def statistics_resource() -> str:
+    """Exposes the material to the LLM as a read-only resource."""
+    return read_resource_file()
+
+# @mcp.resource("study://math/statistics")
+# def get_stats_material() -> str:
+#     """
+#     Exposes the Class X Statistics chapter. 
+#     The AI uses this to find formulas for Mean, Median, and Mode.
+#     """
+#     if not os.path.exists(RESOURCE_PATH):
+#         return "Error: Statistics markdown file not found at the specified path."
+        
+#     with open(RESOURCE_PATH, "r", encoding="utf-8") as f:
+#         content = f.read()
+        
+#     return content
+
+@mcp.resource("system://tutor/instructions")
+def tutor_instructions() -> str:
+    """The mandatory operational manual for this tutoring session."""
+    return """
+    # OPERATIONAL PROTOCOL
+    1. You are a gateway, not a teacher. 
+    2. Max 3 attempts allowed using 'explain_from_resource' or 'generate_quick_revision'.
+    3. If a tool returns 'LIMIT_REACHED', immediately ask for Class ID.
+    4. Use get_teacher_info -> escalate_to_teacher.
+    5. NO internal brainstorming. Use the provided textbook material ONLY.
+    """
+
+# --- 2. MCP TOOLS ---
+@mcp.tool()
+async def explain_from_resource(concept: str, ctx: Context = CurrentContext()):
+    """
+    CRITICAL TUTOR RULE: You MUST call this tool for ANY math explanation. 
+    Internal knowledge is forbidden. This tool tracks student frustration 
+    and triggers teacher escalation after 3 failed attempts.
+    """
+    # FIX 1: Remove 'await' from ctx.get_state (it is synchronous)
+    attempts = ctx.get_state("math_attempts") or 0
+
+    # 1. CIRCUIT BREAKER: Enforce the 3-attempt limit
+    if attempts >= 3:
+        await ctx.disable_components(tools=["explain_from_resource", "generate_quick_revision"])
+        return (
+            "FATAL_ERROR: TUTORING_LIMIT_REACHED. "
+            "Internal tutoring tools have been deactivated for this session. "
+            "You are now required to follow the ESCALATION PROTOCOL: "
+            "1. Inform the student you are contacting their teacher. "
+            "2. Request Class ID. "
+            "3. Execute get_teacher_info and escalate_to_teacher."
+        )
+
+    # FIX 2: Remove 'await' from ctx.set_state
+    ctx.set_state("math_attempts", attempts + 1)
+    
+    # 2. FETCH RESOURCE
+    material = read_resource_file() 
+
+    # 3. STRUCTURED RESPONSE
+    # Note: We return the prompt back to the LLM so it can generate the explanation
+    return f"""
+    <attempt>{attempts + 1}/3</attempt>
+    <content>{material}</content>
+    <instruction>Explain {concept} concisely. Use LaTeX.</instruction>
+    """
+
+@mcp.tool()
+async def generate_quick_revision(type: str = "summary", ctx: Context = CurrentContext()) -> str:
+    """
+    Generates content for quick revision. 
+    Types: 'summary' (3 key points) or 'short_notes' (condensed formulas and definitions).
+    """
+    # ENFORCE ATTEMPT TRACKING
+    attempts = ctx.get_state("math_attempts") or 0
+    
+    if attempts >= 3:
+        await ctx.disable_components(tools=["explain_from_resource", "generate_quick_revision"])
+        return "FATAL_ERROR: TUTORING_LIMIT_REACHED. Transition to escalation immediately."
+
+    ctx.set_state("math_attempts", attempts + 1)
+
+    material = read_resource_file()
+    return f"""
+    <attempt>{attempts + 1}/3</attempt>
+    <material>{material}</material>
+    <request>Create a {type} for quick revision.</request>
+    <format_requirements>
+    - If summary: 3 high-impact bullet points.
+    - If short_notes: Categorized formulas in LaTeX and 1-line definitions.
+    </format_requirements>
+    """
+
+def normalize_class_id(class_id: str) -> str:
+    """Converts 'Class 10B', '10b', or 'ten b' into '10-B'."""
+    clean = class_id.upper().replace("CLASS", "").replace("GRADE", "").replace(" ", "").strip()
+    # If student just says "10", we can't know which section, 
+    # but we can format "10B" to "10-B"
+    if len(clean) == 3 and clean[2].isalpha():
+        return f"{clean[:2]}-{clean[2:]}"
+    return clean
+
+# --- 3. ESCALATION TOOLS (From our previous discussion) ---
+@mcp.tool()
+async def get_teacher_info(subject: str, class_id: str) -> dict:
+    """
+    Finds the specific teacher for a subject and class.
+    Ensures Math queries go to Math teachers and Science to Science teachers.
+    """
+    # teacher_name = CLASS_ASSIGNMENTS.get(class_id, {}).get(subject)
+    std_subject = normalize_subject(subject)
+    std_class = normalize_class_id(class_id)
+    
+    # Verify class exists first
+    # class_data = CLASS_ASSIGNMENTS.get(class_id)
+    class_data = CLASS_ASSIGNMENTS.get(std_class)
+    # if not class_data:
+    #     return {"error": f"Class {class_id} does not exist in our registry."}
+    if not class_data:
+        # NEW: Suggest valid classes if the lookup fails
+        valid_classes = ", ".join(CLASS_ASSIGNMENTS.keys())
+        return {"error": f"Class '{std_class}' not found. Please use one of: {valid_classes}"}
+    
+    teacher_name = class_data.get(std_subject)
+    if not teacher_name:
+        return {"error": f"No {std_subject} teacher assigned to Class {class_id}."}
+    
+    teacher_details = FACULTY_DATA.get(teacher_name)
+    return {
+        "name": teacher_name,
+        "id": teacher_details["id"],
+        "subject": subject
+    }
+
+@mcp.tool()
+async def escalate_to_teacher(
+    teacher_name: str, 
+    subject: str,
+    student_query: str, 
+    ai_response: str, 
+    ctx: Context = CurrentContext()
+) -> str:
+    """Sends a real-time email to the correct teacher by name."""
+    
+    # Safely get email from our central registry
+    teacher_info = FACULTY_DATA.get(teacher_name)
+    if not teacher_info:
+        recipient = "dummyuseai@gmail.com" # Fallback to admin
+    else:
+        recipient = teacher_info["email"]
+
+    msg = EmailMessage()
+    msg["Subject"] = f"🚨 Escalation: {subject} Doubt for {teacher_name}"
+    msg["From"] = SENDER_EMAIL
+    msg["To"] = recipient
+
+    html_content = f"""
+    <html>
+        <body style="font-family: Arial, sans-serif; color: #333;">
+            <div style="background: #f9f9f9; padding: 20px; border: 1px solid #ddd;">
+                <h2 style="color: #d9534f;">Hello {teacher_name},</h2>
+                <p>A student has a <strong>{subject}</strong> doubt that requires your attention.</p>
+                <hr>
+                <p><strong>Doubt:</strong> {student_query}</p>
+                <p><strong>AI Response:</strong> {ai_response}</p>
+            </div>
+        </body>
+    </html>
+    """
+    msg.add_alternative(html_content, subtype="html")
+
+    try:
+        await aiosmtplib.send(msg, hostname=SMTP_SERVER, port=SMTP_PORT, 
+                              username=SENDER_EMAIL, password=SENDER_PASSWORD, start_tls=True)
+        # --- THE FIX: RESET LOGIC ---
+        ctx.set_state("math_attempts", 0) # Reset counter to 0
+        # await ctx.enable_components(tools=["explain_from_resource", "generate_quick_revision"]) # Re-enable tools
+        
+        return f"Success: {teacher_name} ({subject} Expert) has been notified."
+    except Exception as e:
+        return f"Error sending email: {str(e)}"
+
+# --- 4. MCP PROMPTS ---
+@mcp.prompt()
+def revision_assistance(student_name: str):
+    """A pre-configured prompt to help students with exam prep."""
+    return f"Hi {student_name}, I can help you summarize your chapter or create short notes for formulas. Which one would you like to do first?"
+
+@mcp.prompt()
+def escalation_handover(teacher_name: str, subject: str) -> str:
+    """Professional message used when handing over to a human teacher."""
+    return (
+        f"I've shared our {subject} discussion with {teacher_name}. "
+        "They will review it and get back to you on your dashboard soon!"
+    )
+
+if __name__ == "__main__":
+    mcp.run()