Update server.py

server.py

@@ -182,42 +182,115 @@ async def process_text(file_bytes: bytes) -> str:
     except UnicodeDecodeError:
         return "Error: File is not valid UTF-8 text"
 
+def decode_file(file_data: str | bytes) -> bytes:
+    """Decode base64 file data to bytes"""
+    if isinstance(file_data, bytes):
+        return file_data
+    
+    # If it's a string, decode from base64
+    try:
+        return base64.b64decode(file_data)
+    except Exception as e:
+        raise ValueError(f"Failed to decode base64: {e}")
+
+def detect_file_type(file_bytes: bytes) -> str:
+    """Detect file type from magic bytes"""
+    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"  # GIF
+    elif file_bytes.startswith(b'\x42\x4d'):
+        return "image"  # BMP
+    else:
+        # Try to decode as text
+        try:
+            file_bytes[:1000].decode('utf-8')
+            return "text"
+        except:
+            return "unknown"
+
+async def process_pdf(file_bytes: bytes) -> str:
+    """Process PDF file and extract text/info"""
+    try:
+        pdf_file = io.BytesIO(file_bytes)
+        reader = PdfReader(pdf_file)
+        
+        num_pages = len(reader.pages)
+        text = ""
+        for page in reader.pages:
+            text += page.extract_text()
+        
+        return f"PDF Analysis:\n- Pages: {num_pages}\n- Text length: {len(text)} chars\n\nFirst 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:
+    """Process image file"""
+    try:
+        img = Image.open(io.BytesIO(file_bytes))
+        return f"Image Analysis:\n- Format: {img.format}\n- Size: {img.size[0]}x{img.size[1]} pixels\n- Mode: {img.mode}"
+    except Exception as e:
+        return f"Image processing failed: {str(e)}"
+
+async def process_text(file_bytes: bytes) -> str:
+    """Process text file"""
+    try:
+        text = file_bytes.decode('utf-8')
+        lines = text.split('\n')
+        words = len(text.split())
+        return f"Text Analysis:\n- Lines: {len(lines)}\n- Words: {words}\n- Characters: {len(text)}\n\nFirst 500 chars:\n{text[:500]}"
+    except UnicodeDecodeError:
+        return "Error: File is not valid UTF-8 text"
+
 @mcp.tool 
 async def analyze_file(
     ctx: Context, 
-    file: bytes,
+    file: str,  # MCP sends base64-encoded string
     file_type: Optional[Literal["text", "image", "pdf"]] = None
 ) -> str: 
     """
-    Analyzes a file.
+    Analyzes a file with auto-detection.
     
     Args:
-        file: The file content
-        file_type: Type of file (text/image/pdf). If not provided, will ask user.
+        file: The file content (base64-encoded)
+        file_type: Optional. If not provided, will auto-detect.
     """
-    # Basic info
-    info = f"Received file of size {len(file)} bytes.\n\n"
+    # Decode base64 to bytes
+    try:
+        file_bytes = decode_file(file)
+    except ValueError as e:
+        return f"Error decoding file: {e}"
     
-    # If file_type not provided, elicit it from user
+    info = f"Received file of size {len(file_bytes)} bytes.\n"
+    
+    # Auto-detect if not provided
     if file_type is None:
-        result = await ctx.elicit(
-            message="What is the file type?",
-            response_type=["text", "image", "pdf"]
-        )
+        file_type = detect_file_type(file_bytes)
+        info += f"Auto-detected type: {file_type}\n\n"
         
-        if result.action == "accept":
-            file_type = result.data
-        else:
-            return "Elicitation cancelled by user"
+        if file_type == "unknown":
+            return info + "Could not detect file type. Please call again with file_type parameter (text/image/pdf)"
+    else:
+        # Verify the file_type matches actual content
+        detected = detect_file_type(file_bytes)
+        info += f"Specified type: {file_type}\n"
+        info += f"Detected type: {detected}\n\n"
+        
+        if detected != file_type and detected != "unknown":
+            info += f"⚠️ Warning: Specified type '{file_type}' doesn't match detected type '{detected}'\n\n"
     
-    # Route to appropriate processor
+    # Process based on file_type
     try:
         if file_type == "pdf":
-            result = await process_pdf(file)
+            result = await process_pdf(file_bytes)
         elif file_type == "image":
-            result = await process_image(file)
+            result = await process_image(file_bytes)
         elif file_type == "text":
-            result = await process_text(file)
+            result = await process_text(file_bytes)
         else:
             return info + f"Unknown file type: {file_type}"