Fix MCP server - use FastAPI instead of MCP SDK for Hugging Face Spaces

app.py

@@ -8,20 +8,28 @@ import json
 import asyncio
 import logging
 from typing import Any, Dict, List, Optional
-from mcp.server import Server
-from mcp.server.models import InitializationOptions
-from mcp.server.sse import sse_server
-from mcp.types import (
-    Resource,
-    Tool,
-    TextContent,
-    LoggingLevel
-)
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+import uvicorn
 
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
+# Initialize FastAPI app
+app = FastAPI(title="MCP Documentation Server", version="1.0.0")
+
+# Add CORS middleware
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
 # Global data storage
 chunks_data = None
 docs_data = None
@@ -45,56 +53,58 @@ def load_data():
         logger.error(f"Error loading data: {e}")
         raise
 
-# Initialize the MCP server
-server = Server("mcp-docs-server")
+# Pydantic models
+class SearchRequest(BaseModel):
+    query: str
+    limit: int = 5
 
-@server.list_resources()
-async def list_resources() -> List[Resource]:
-    """List available documentation resources"""
-    if not docs_data:
-        return []
-    
-    resources = []
-    for doc in docs_data:
-        resources.append(Resource(
-            uri=f"mcp://docs/{doc.get('id', 'unknown')}",
-            name=doc.get('title', 'Untitled'),
-            description=doc.get('content', '')[:200] + "..." if len(doc.get('content', '')) > 200 else doc.get('content', ''),
-            mimeType="text/plain"
-        ))
-    
-    return resources
+class SearchResponse(BaseModel):
+    results: List[Dict[str, Any]]
+    total: int
 
-@server.read_resource()
-async def read_resource(uri: str) -> str:
-    """Read a specific documentation resource"""
-    if not chunks_data:
-        return "Data not loaded"
-    
-    # Extract document ID from URI
-    if uri.startswith("mcp://docs/"):
-        doc_id = uri.replace("mcp://docs/", "")
-        
-        # Find chunks for this document
-        doc_chunks = [chunk for chunk in chunks_data if chunk.get('doc_id') == doc_id]
-        
-        if doc_chunks:
-            # Combine all chunks for the document
-            content = "\n\n".join([chunk.get('text', '') for chunk in doc_chunks])
-            return content
-        else:
-            return f"Document {doc_id} not found"
-    
-    return "Invalid URI"
+class ToolCallRequest(BaseModel):
+    name: str
+    arguments: Dict[str, Any]
 
-@server.list_tools()
-async def list_tools() -> List[Tool]:
-    """List available tools"""
+class ToolCallResponse(BaseModel):
+    content: List[Dict[str, str]]
+
+@app.on_event("startup")
+async def startup_event():
+    """Load data on startup"""
+    load_data()
+
+@app.get("/")
+async def root():
+    """Health check endpoint"""
+    return {
+        "message": "MCP Documentation Server",
+        "status": "running",
+        "chunks_loaded": len(chunks_data) if chunks_data else 0,
+        "docs_loaded": len(docs_data) if docs_data else 0,
+        "mcp_server": True
+    }
+
+@app.get("/mcp/info")
+async def mcp_info():
+    """MCP server information"""
+    return {
+        "name": "mcp-docs-server",
+        "version": "1.0.0",
+        "capabilities": {
+            "tools": True,
+            "resources": True
+        }
+    }
+
+@app.get("/mcp/tools")
+async def list_tools():
+    """List available MCP tools"""
     return [
-        Tool(
-            name="search_docs",
-            description="Search through MCP documentation chunks",
-            inputSchema={
+        {
+            "name": "search_docs",
+            "description": "Search through MCP documentation chunks",
+            "inputSchema": {
                 "type": "object",
                 "properties": {
                     "query": {
@@ -109,11 +119,11 @@ async def list_tools() -> List[Tool]:
                 },
                 "required": ["query"]
             }
-        ),
-        Tool(
-            name="get_chunk",
-            description="Get a specific documentation chunk by ID",
-            inputSchema={
+        },
+        {
+            "name": "get_chunk",
+            "description": "Get a specific documentation chunk by ID",
+            "inputSchema": {
                 "type": "object",
                 "properties": {
                     "chunk_id": {
@@ -123,22 +133,25 @@ async def list_tools() -> List[Tool]:
                 },
                 "required": ["chunk_id"]
             }
-        ),
-        Tool(
-            name="list_docs",
-            description="List all available documents",
-            inputSchema={
+        },
+        {
+            "name": "list_docs",
+            "description": "List all available documents",
+            "inputSchema": {
                 "type": "object",
                 "properties": {}
             }
-        )
+        }
     ]
 
-@server.call_tool()
-async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
-    """Handle tool calls"""
+@app.post("/mcp/tools/call")
+async def call_tool(request: ToolCallRequest):
+    """Call an MCP tool"""
     if not chunks_data:
-        return [TextContent(type="text", text="Data not loaded")]
+        raise HTTPException(status_code=500, detail="Data not loaded")
+    
+    name = request.name
+    arguments = request.arguments
     
     if name == "search_docs":
         query = arguments.get("query", "").lower()
@@ -178,7 +191,7 @@ async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
         else:
             response = f"No results found for '{arguments.get('query', '')}'"
         
-        return [TextContent(type="text", text=response)]
+        return ToolCallResponse(content=[{"type": "text", "text": response}])
     
     elif name == "get_chunk":
         chunk_id = arguments.get("chunk_id", "")
@@ -189,13 +202,13 @@ async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
                 response += f"{chunk.get('text', '')}\n\n"
                 response += f"Source: {chunk.get('filename', 'Unknown')}\n"
                 response += f"URL: {chunk.get('url', 'N/A')}"
-                return [TextContent(type="text", text=response)]
+                return ToolCallResponse(content=[{"type": "text", "text": response}])
         
-        return [TextContent(type="text", text=f"Chunk {chunk_id} not found")]
+        return ToolCallResponse(content=[{"type": "text", "text": f"Chunk {chunk_id} not found"}])
     
     elif name == "list_docs":
         if not docs_data:
-            return [TextContent(type="text", text="No documents available")]
+            return ToolCallResponse(content=[{"type": "text", "text": "No documents available"}])
         
         response = "Available documents:\n\n"
         for doc in docs_data:
@@ -203,30 +216,108 @@ async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
             response += f"  ID: {doc.get('id', 'Unknown')}\n"
             response += f"  URL: {doc.get('url', 'N/A')}\n\n"
         
-        return [TextContent(type="text", text=response)]
+        return ToolCallResponse(content=[{"type": "text", "text": response}])
     
     else:
-        return [TextContent(type="text", text=f"Unknown tool: {name}")]
+        return ToolCallResponse(content=[{"type": "text", "text": f"Unknown tool: {name}"}])
 
-async def main():
-    """Main entry point"""
-    # Load data
-    load_data()
+@app.get("/mcp/resources")
+async def list_resources():
+    """List available MCP resources"""
+    if not docs_data:
+        return []
+    
+    resources = []
+    for doc in docs_data:
+        resources.append({
+            "uri": f"mcp://docs/{doc.get('id', 'unknown')}",
+            "name": doc.get('title', 'Untitled'),
+            "description": doc.get('content', '')[:200] + "..." if len(doc.get('content', '')) > 200 else doc.get('content', ''),
+            "mimeType": "text/plain"
+        })
+    
+    return resources
+
+@app.get("/mcp/resources/{resource_id}")
+async def read_resource(resource_id: str):
+    """Read a specific MCP resource"""
+    if not chunks_data:
+        return "Data not loaded"
+    
+    # Find chunks for this document
+    doc_chunks = [chunk for chunk in chunks_data if chunk.get('doc_id') == resource_id]
     
-    # Run the server with SSE transport for HTTP access
-    async with sse_server() as (read_stream, write_stream):
-        await server.run(
-            read_stream,
-            write_stream,
-            InitializationOptions(
-                server_name="mcp-docs-server",
-                server_version="1.0.0",
-                capabilities=server.get_capabilities(
-                    notification_options=None,
-                    experimental_capabilities=None
-                )
-            )
+    if doc_chunks:
+        # Combine all chunks for the document
+        content = "\n\n".join([chunk.get('text', '') for chunk in doc_chunks])
+        return content
+    else:
+        return f"Document {resource_id} not found"
+
+# Legacy REST API endpoints for backward compatibility
+@app.post("/search", response_model=SearchResponse)
+async def search_docs(request: SearchRequest):
+    """Search through documentation chunks using text matching"""
+    if not chunks_data:
+        raise HTTPException(status_code=500, detail="Data not loaded")
+    
+    try:
+        query_lower = request.query.lower()
+        results = []
+        
+        for chunk in chunks_data:
+            text = chunk.get('text', '').lower()
+            title = chunk.get('title', '').lower()
+            
+            # Simple scoring based on query matches
+            score = 0
+            if query_lower in text:
+                score += text.count(query_lower) * 2  # Text matches worth more
+            if query_lower in title:
+                score += title.count(query_lower) * 5  # Title matches worth most
+            
+            if score > 0:
+                results.append({
+                    "chunk_id": chunk.get('chunk_id'),
+                    "title": chunk.get('title'),
+                    "text": chunk.get('text'),
+                    "url": chunk.get('url'),
+                    "filename": chunk.get('filename'),
+                    "chunk_index": chunk.get('chunk_index'),
+                    "total_chunks": chunk.get('total_chunks'),
+                    "score": score
+                })
+        
+        # Sort by relevance score
+        results = sorted(results, key=lambda x: x['score'], reverse=True)
+        
+        return SearchResponse(
+            results=results[:request.limit],
+            total=len(results)
         )
+        
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Search error: {str(e)}")
+
+@app.get("/chunks/{chunk_id}")
+async def get_chunk(chunk_id: str):
+    """Get a specific chunk by ID"""
+    if not chunks_data:
+        raise HTTPException(status_code=500, detail="Data not loaded")
+    
+    for chunk in chunks_data:
+        if chunk.get('chunk_id') == chunk_id:
+            return chunk
+    
+    raise HTTPException(status_code=404, detail="Chunk not found")
+
+@app.get("/docs")
+async def list_docs():
+    """List all available documents"""
+    if not docs_data:
+        raise HTTPException(status_code=500, detail="Data not loaded")
+    
+    return {"documents": docs_data}
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    uvicorn.run(app, host="0.0.0.0", port=7860)

requirements.txt

@@ -1 +1,4 @@
-mcp==1.0.0
+fastapi==0.104.1
+uvicorn==0.24.0
+pydantic==2.5.0
+python-multipart==0.0.6

test_mcp_connection.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+"""
+Test script to verify MCP server connection
+"""
+
+import asyncio
+import aiohttp
+import json
+
+async def test_mcp_server():
+    """Test the MCP server on Hugging Face Spaces"""
+    url = "https://galcan-mcp-docs-server.hf.space"
+    
+    print("Testing MCP server connection...")
+    print(f"URL: {url}")
+    
+    try:
+        async with aiohttp.ClientSession() as session:
+            # Test health check
+            async with session.get(f"{url}/") as response:
+                if response.status == 200:
+                    data = await response.json()
+                    print(f"[OK] Health check passed: {data}")
+                else:
+                    print(f"[ERROR] Health check failed: {response.status}")
+                    return False
+            
+            # Test search endpoint
+            search_data = {
+                "query": "MCP architecture",
+                "limit": 3
+            }
+            
+            async with session.post(f"{url}/search", json=search_data) as response:
+                if response.status == 200:
+                    data = await response.json()
+                    print(f"[OK] Search test passed: Found {data.get('total', 0)} results")
+                else:
+                    print(f"[ERROR] Search test failed: {response.status}")
+                    return False
+            
+            print("[SUCCESS] MCP server is working correctly!")
+            return True
+            
+    except Exception as e:
+        print(f"[ERROR] Connection failed: {e}")
+        return False
+
+if __name__ == "__main__":
+    asyncio.run(test_mcp_server())