#!/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 @app.get("/", response_class=HTMLResponse) async def root(): """Root endpoint with server information""" return f""" MCP Server - Hugging Face Spaces

🤖 MCP Server

Welcome to the Model Context Protocol server!

Tools Available: {len(server.tools)}

Status: ✅ Running

📚 View API Documentation

""" # Health check endpoint @app.get("/health", response_model=HealthResponse) 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 @app.post("/mcp", response_model=MCPResponse) 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 @app.get("/tools") 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 @app.get("/resources") 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 @app.post("/tools/{tool_name}") 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()