Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| HTTP wrapper for the MCP server to run on Hugging Face Spaces | |
| """ | |
| from fastapi import FastAPI, HTTPException, Request | |
| from fastapi.responses import HTMLResponse | |
| from pydantic import BaseModel | |
| import asyncio | |
| import json | |
| import logging | |
| import os | |
| from typing import Any, Dict, Optional | |
| from datetime import datetime | |
| import uvicorn | |
| # Import our MCP server | |
| from mcp_server import server | |
| # Setup logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # FastAPI app | |
| app = FastAPI( | |
| title="MCP Server HTTP Wrapper", | |
| description="HTTP interface for Model Context Protocol server", | |
| version="1.0.0", | |
| docs_url="/docs", | |
| redoc_url="/redoc" | |
| ) | |
| # Request/Response models | |
| class MCPRequest(BaseModel): | |
| method: str | |
| params: Dict[str, Any] = {} | |
| id: Optional[int] = 1 | |
| class MCPResponse(BaseModel): | |
| jsonrpc: str = "2.0" | |
| id: Optional[int] = None | |
| result: Optional[Any] = None | |
| error: Optional[Dict[str, Any]] = None | |
| class HealthResponse(BaseModel): | |
| status: str | |
| timestamp: str | |
| server_name: str | |
| version: str | |
| tools_count: int | |
| resources_count: int | |
| # Root endpoint with basic information | |
| async def root(): | |
| """Root endpoint with server information""" | |
| return f""" | |
| <!DOCTYPE html> | |
| <html> | |
| <head> | |
| <title>MCP Server - Hugging Face Spaces</title> | |
| <style> | |
| body {{ font-family: Arial, sans-serif; margin: 40px; background-color: #f5f5f5; }} | |
| .container {{ max-width: 800px; margin: 0 auto; background: white; padding: 30px; border-radius: 10px; }} | |
| h1 {{ color: #333; border-bottom: 2px solid #007acc; padding-bottom: 10px; }} | |
| </style> | |
| </head> | |
| <body> | |
| <div class="container"> | |
| <h1>🤖 MCP Server</h1> | |
| <p>Welcome to the Model Context Protocol server!</p> | |
| <p><strong>Tools Available:</strong> {len(server.tools)}</p> | |
| <p><strong>Status:</strong> ✅ Running</p> | |
| <p><a href="/docs">📚 View API Documentation</a></p> | |
| </div> | |
| </body> | |
| </html> | |
| """ | |
| # Health check endpoint | |
| async def health_check(): | |
| """Health check endpoint""" | |
| return HealthResponse( | |
| status="healthy", | |
| timestamp=datetime.now().isoformat(), | |
| server_name=server.name, | |
| version=server.version, | |
| tools_count=len(server.tools), | |
| resources_count=len(server.resources) | |
| ) | |
| # Main MCP endpoint | |
| async def mcp_endpoint(request: MCPRequest): | |
| """Main MCP endpoint for handling requests""" | |
| try: | |
| logger.info(f"Received MCP request: {request.method}") | |
| # Convert Pydantic model to dict for the MCP server | |
| mcp_request = { | |
| "jsonrpc": "2.0", | |
| "id": request.id, | |
| "method": request.method, | |
| "params": request.params | |
| } | |
| # Process the request through our MCP server | |
| response = await server.handle_request(mcp_request) | |
| # Convert response to our response model | |
| return MCPResponse( | |
| jsonrpc=response.get("jsonrpc", "2.0"), | |
| id=response.get("id"), | |
| result=response.get("result"), | |
| error=response.get("error") | |
| ) | |
| except Exception as e: | |
| logger.error(f"Error processing MCP request: {e}") | |
| return MCPResponse( | |
| jsonrpc="2.0", | |
| id=request.id, | |
| error={ | |
| "code": -32603, | |
| "message": f"Internal error: {str(e)}" | |
| } | |
| ) | |
| # Convenience endpoints | |
| async def list_tools(): | |
| """List all available tools""" | |
| request = MCPRequest(method="tools/list") | |
| response = await mcp_endpoint(request) | |
| return response.result if response.result else response.error | |
| async def list_resources(): | |
| """List all available resources""" | |
| request = MCPRequest(method="resources/list") | |
| response = await mcp_endpoint(request) | |
| return response.result if response.result else response.error | |
| async def execute_tool(tool_name: str, arguments: Dict[str, Any] = None): | |
| """Execute a specific tool with arguments""" | |
| if arguments is None: | |
| arguments = {} | |
| request = MCPRequest( | |
| method="tools/call", | |
| params={ | |
| "name": tool_name, | |
| "arguments": arguments | |
| } | |
| ) | |
| return await mcp_endpoint(request) | |
| # Main function | |
| def main(): | |
| """Main function to run the HTTP server""" | |
| port = int(os.getenv("PORT", 7860)) | |
| host = os.getenv("HOST", "0.0.0.0") | |
| logger.info(f"Starting HTTP server on {host}:{port}") | |
| uvicorn.run( | |
| app, | |
| host=host, | |
| port=port, | |
| log_level="info", | |
| access_log=True | |
| ) | |
| if __name__ == "__main__": | |
| main() | |