server.py

from fastmcp import FastMCP
from fastmcp import FastMCP, Context
from fastmcp.server.elicitation import AcceptedElicitation
import logging
import base64
from typing import Literal, Optional
import io
from PyPDF2 import PdfReader
from PIL import Image
import os

# Initialize the server
mcp = FastMCP("Math-Education-Server")

# SENIOR TIP: Use absolute paths relative to the script location 
# This prevents "File Not Found" errors in Docker.
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
MARKDOWN_FILE = os.path.join(CURRENT_DIR, "resource", "jemh114-min (1).md")

def get_local_resource():
    if os.path.exists(MARKDOWN_FILE):
        with open(MARKDOWN_FILE, "r", encoding="utf-8") as f:
            return f.read()
    return "Resource file not found."

@mcp.tool()
async def generate_chapter_summary(chapter_name: str) -> str:
    """
    Provides source material + strict instructions for a math chapter summary.
    """
    raw_data = get_local_resource()

    return f"""
You are a Mathematics Expert and Academic Tutor.

TASK:
Generate a structured summary of the chapter titled: "{chapter_name}"

IMPORTANT RULES:
- Use ONLY the provided <source_material>.
- If something is missing in the material, explicitly write: "Not found in source".
- Do NOT invent formulas, theorems, or examples.
- Prefer mathematical notation and LaTeX formatting.
- Output must be clean Markdown.

OUTPUT FORMAT (STRICT):
## Chapter: {chapter_name}

### 1. 3-Line High-Level Summary (3 bullets max)
- ...
- ...
- ...

### 2. Key Concepts (Definitions / Theorems / Properties)
List important definitions and results.
Each bullet must follow:
- **Concept Name**: Explanation (1–2 lines)

### 3. Important Formulas (LaTeX only)
Provide all important formulas mentioned in the chapter.
Format each as:
- **Formula Name**: $...$

### 4. Step-by-Step Core Method(s)
If the chapter contains any algorithm/method (example: solving quadratic, integration steps),
list the procedure in numbered steps.

### 5. Common Mistakes / Traps
List 3–6 common mistakes students make.

### 6. Quick Worked Example (If present in source)
- Problem:
- Solution (brief but clear):
- Final Answer:

If no example exists in the source, write: "Not found in source".

SOURCE:
<source_material>
{raw_data[:4000]}
</source_material>
"""


@mcp.tool()
async def generate_quiz(chapter_name: str, difficulty: str = "medium") -> str:
    """
    Generates quiz questions based on chapter material.
    """
    raw_data = get_local_resource()

    return f"""
You are a Mathematics Professor designing an exam.

CHAPTER: "{chapter_name}"
DIFFICULTY: "{difficulty}"

DIFFICULTY RULES:
- easy: direct definition/formula substitution questions
- medium: multi-step numeric problems + conceptual reasoning
- hard: tricky reasoning, proofs, edge cases, mixed-concept questions

IMPORTANT RULES:
- Use ONLY the provided source material.
- Do NOT invent topics not present in the material.
- Every question must match the difficulty level.
- Provide answer key + short solution steps.
- Format must be clean Markdown.
- Use LaTeX for all formulas.

OUTPUT FORMAT (STRICT):

## Quiz: {chapter_name} ({difficulty})

### Q1. (Conceptual / Definition Based)
Question text...

**Options (if MCQ):**
A) ...
B) ...
C) ...
D) ...

### Q2. (Formula/Application Based)
Question text...

### Q3. (Multi-step Problem / Reasoning Based)
Question text...

---

## Answer Key + Solutions

### Q1 Solution
- Correct Answer: ...
- Explanation: ...

### Q2 Solution
Step 1: ...
Step 2: ...
Final Answer: ...

### Q3 Solution
Step 1: ...
Step 2: ...
Step 3: ...
Final Answer: ...

SOURCE:
<source_material>
{raw_data[:4000]}
</source_material>
"""
    
# Helper functions for different file types
async def process_pdf(file_bytes: bytes) -> str:
    try:
        pdf_file = io.BytesIO(file_bytes)
        reader = PdfReader(pdf_file)

        num_pages = len(reader.pages)
        text = ""
        for page in reader.pages:
            extracted = page.extract_text() or ""
            text += extracted

        return (
            f"PDF Analysis:\n"
            f"- Pages: {num_pages}\n"
            f"- Text length: {len(text)} chars\n\n"
            f"First 500 chars:\n{text[:500]}"
        )
    except Exception as e:
        return f"PDF processing failed: {str(e)}\nThis might not be a valid PDF file."


async def process_image(file_bytes: bytes) -> str:
    try:
        img = Image.open(io.BytesIO(file_bytes))
        return (
            f"Image Analysis:\n"
            f"- Format: {img.format}\n"
            f"- Size: {img.size[0]}x{img.size[1]} pixels\n"
            f"- Mode: {img.mode}"
        )
    except Exception as e:
        return f"Image processing failed: {str(e)}"


async def process_text(file_bytes: bytes) -> str:
    try:
        text = file_bytes.decode("utf-8")
        lines = text.split("\n")
        words = len(text.split())

        return (
            f"Text Analysis:\n"
            f"- Lines: {len(lines)}\n"
            f"- Words: {words}\n"
            f"- Characters: {len(text)}\n\n"
            f"First 500 chars:\n{text[:500]}"
        )
    except UnicodeDecodeError:
        return "Error: File is not valid UTF-8 text"


def detect_file_type(file_bytes: bytes) -> str:
    if file_bytes.startswith(b"%PDF"):
        return "pdf"
    elif file_bytes.startswith(b"\x89PNG"):
        return "image"
    elif file_bytes.startswith(b"\xff\xd8\xff"):
        return "image"  # JPEG
    elif file_bytes.startswith(b"GIF87a") or file_bytes.startswith(b"GIF89a"):
        return "image"
    elif file_bytes.startswith(b"\x42\x4d"):
        return "image"  # BMP
    else:
        try:
            file_bytes[:1000].decode("utf-8")
            return "text"
        except:
            return "unknown"


# ---------------- MCP TOOL ---------------- #

@mcp.tool
async def analyze_file(
    ctx: Context,
    file_uri: str,  # <-- file path/uri from client
    file_type: Optional[Literal["text", "image", "pdf"]] = None,
) -> str:
    """
    Analyze a file using MCP resource URI (recommended).

    file_uri examples:
      - file:///C:/Users/Admin/Desktop/sample.pdf
      - resource://uploads/123/sample.pdf
    """

    # Read file bytes via MCP resource API
    try:
        resource_result = await ctx.read_resource(file_uri)
    except Exception as e:
        return f"Failed to read resource '{file_uri}': {str(e)}"

    if not resource_result.contents:
        return f"No content returned for resource: {file_uri}"

    content = resource_result.contents[0]

    # If it's text resource
    if hasattr(content, "text") and content.text is not None:
        file_bytes = content.text.encode("utf-8")
    else:
        # Blob resource (bytes might be base64 encoded depending on client)
        blob_data = content.blob

        # some clients give blob as base64 string
        if isinstance(blob_data, str):
            file_bytes = base64.b64decode(blob_data)
        else:
            file_bytes = blob_data

    info = f"Loaded file from URI: {file_uri}\nSize: {len(file_bytes)} bytes\n"

    # Auto detect if not specified
    if file_type is None:
        detected = detect_file_type(file_bytes)
        info += f"Auto-detected type: {detected}\n\n"

        if detected == "unknown":
            return info + "Could not detect file type. Please specify file_type=text/image/pdf"

        file_type = detected
    else:
        detected = detect_file_type(file_bytes)
        info += f"Specified type: {file_type}\nDetected type: {detected}\n\n"

        if detected != file_type and detected != "unknown":
            info += f"⚠️ Warning: mismatch (specified={file_type}, detected={detected})\n\n"

    # Process file
    if file_type == "pdf":
        result = await process_pdf(file_bytes)
    elif file_type == "image":
        result = await process_image(file_bytes)
    elif file_type == "text":
        result = await process_text(file_bytes)
    else:
        return info + f"Unsupported file_type: {file_type}"

    return info + result

        
if __name__ == "__main__":
    print("Starting FastMCP server on port 7860...")
    
    # We use the 'sse' transport. This is crucial for the "handshake".
    # Standard 'http' can sometimes cause issues with clients expecting
    # a persistent event stream for the initialization.
    mcp.run(transport="http", host="0.0.0.0", port=7860)