setup_script.sh

#!/bin/bash

# Setup script for HF MCP Server project
# Run this script to create the project structure on your local machine

set -e  # Exit on any error

# Configuration
PROJECT_DIR="/Volumes/TOSHIBA_EXT/huggingface/hf-mcp-server"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

echo "🚀 Setting up HF MCP Server project..."
echo "Target directory: $PROJECT_DIR"

# Create project directory
echo "📁 Creating project directory..."
mkdir -p "$PROJECT_DIR"
cd "$PROJECT_DIR"

# Create main Python files
echo "📝 Creating mcp_server.py..."
cat > mcp_server.py << 'EOF'
#!/usr/bin/env python3
"""
MCP Server with sample tools for Hugging Face Spaces deployment
"""

import asyncio
import json
import logging
import sys
from typing import Any, Dict, List, Optional, Union
from datetime import datetime
import os

# MCP Server implementation
class MCPServer:
    def __init__(self, name: str = "hf-mcp-server", version: str = "1.0.0"):
        self.name = name
        self.version = version
        self.tools = {}
        self.resources = {}
        self.setup_logging()
        
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(self.name)
    
    def tool(self, name: str, description: str = "", parameters: Dict = None):
        """Decorator to register tools"""
        def decorator(func):
            self.tools[name] = {
                "function": func,
                "description": description,
                "parameters": parameters or {}
            }
            return func
        return decorator
    
    def resource(self, uri: str, name: str = "", description: str = ""):
        """Decorator to register resources"""
        def decorator(func):
            self.resources[uri] = {
                "function": func,
                "name": name,
                "description": description
            }
            return func
        return decorator
    
    async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Handle incoming MCP requests"""
        try:
            method = request.get("method")
            params = request.get("params", {})
            request_id = request.get("id", 1)
            
            if method == "initialize":
                return await self.handle_initialize(request_id, params)
            elif method == "tools/list":
                return await self.handle_list_tools(request_id)
            elif method == "tools/call":
                return await self.handle_call_tool(request_id, params)
            elif method == "resources/list":
                return await self.handle_list_resources(request_id)
            elif method == "resources/read":
                return await self.handle_read_resource(request_id, params)
            else:
                return {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "error": {
                        "code": -32601,
                        "message": f"Method not found: {method}"
                    }
                }
        except Exception as e:
            self.logger.error(f"Error handling request: {e}")
            return {
                "jsonrpc": "2.0",
                "id": request.get("id", 1),
                "error": {
                    "code": -32603,
                    "message": f"Internal error: {str(e)}"
                }
            }
    
    async def handle_initialize(self, request_id: int, params: Dict) -> Dict:
        """Handle initialization request"""
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {
                    "tools": {},
                    "resources": {}
                },
                "serverInfo": {
                    "name": self.name,
                    "version": self.version
                }
            }
        }
    
    async def handle_list_tools(self, request_id: int) -> Dict:
        """Handle tools list request"""
        tools_list = []
        for name, tool_info in self.tools.items():
            tools_list.append({
                "name": name,
                "description": tool_info["description"],
                "inputSchema": {
                    "type": "object",
                    "properties": tool_info["parameters"],
                    "required": list(tool_info["parameters"].keys()) if tool_info["parameters"] else []
                }
            })
        
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "tools": tools_list
            }
        }
    
    async def handle_call_tool(self, request_id: int, params: Dict) -> Dict:
        """Handle tool call request"""
        tool_name = params.get("name")
        arguments = params.get("arguments", {})
        
        if tool_name not in self.tools:
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {
                    "code": -32602,
                    "message": f"Tool not found: {tool_name}"
                }
            }
        
        try:
            tool_func = self.tools[tool_name]["function"]
            result = await tool_func(**arguments)
            
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "content": [
                        {
                            "type": "text",
                            "text": str(result)
                        }
                    ]
                }
            }
        except Exception as e:
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {
                    "code": -32603,
                    "message": f"Tool execution error: {str(e)}"
                }
            }
    
    async def handle_list_resources(self, request_id: int) -> Dict:
        """Handle resources list request"""
        resources_list = []
        for uri, resource_info in self.resources.items():
            resources_list.append({
                "uri": uri,
                "name": resource_info["name"],
                "description": resource_info["description"],
                "mimeType": "text/plain"
            })
        
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "resources": resources_list
            }
        }
    
    async def handle_read_resource(self, request_id: int, params: Dict) -> Dict:
        """Handle resource read request"""
        uri = params.get("uri")
        
        if uri not in self.resources:
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {
                    "code": -32602,
                    "message": f"Resource not found: {uri}"
                }
            }
        
        try:
            resource_func = self.resources[uri]["function"]
            content = await resource_func()
            
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "contents": [
                        {
                            "uri": uri,
                            "mimeType": "text/plain",
                            "text": str(content)
                        }
                    ]
                }
            }
        except Exception as e:
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "error": {
                    "code": -32603,
                    "message": f"Resource read error: {str(e)}"
                }
            }


# Create server instance
server = MCPServer("hf-mcp-server", "1.0.0")


# Sample Tools
@server.tool(
    name="echo",
    description="Echo back the input text",
    parameters={
        "text": {
            "type": "string",
            "description": "Text to echo back"
        }
    }
)
async def echo_tool(text: str) -> str:
    """Simple echo tool"""
    return f"Echo: {text}"


@server.tool(
    name="current_time",
    description="Get the current timestamp",
    parameters={}
)
async def current_time_tool() -> str:
    """Get current time"""
    return f"Current time: {datetime.now().isoformat()}"


@server.tool(
    name="calculate",
    description="Perform basic mathematical calculations",
    parameters={
        "expression": {
            "type": "string",
            "description": "Mathematical expression to evaluate (e.g., '2 + 2')"
        }
    }
)
async def calculate_tool(expression: str) -> str:
    """Simple calculator tool"""
    try:
        # Basic safety check - only allow certain characters
        allowed_chars = set("0123456789+-*/()%. ")
        if not all(c in allowed_chars for c in expression):
            return "Error: Invalid characters in expression"
        
        result = eval(expression)
        return f"Result: {expression} = {result}"
    except Exception as e:
        return f"Error: {str(e)}"


@server.tool(
    name="word_count",
    description="Count words in the given text",
    parameters={
        "text": {
            "type": "string",
            "description": "Text to count words in"
        }
    }
)
async def word_count_tool(text: str) -> str:
    """Count words in text"""
    words = text.split()
    chars = len(text)
    chars_no_spaces = len(text.replace(" ", ""))
    
    return f"Text analysis:\n- Words: {len(words)}\n- Characters: {chars}\n- Characters (no spaces): {chars_no_spaces}"


@server.tool(
    name="reverse_text",
    description="Reverse the given text",
    parameters={
        "text": {
            "type": "string",
            "description": "Text to reverse"
        }
    }
)
async def reverse_text_tool(text: str) -> str:
    """Reverse text"""
    return f"Reversed: {text[::-1]}"


# Sample Resources
@server.resource(
    uri="server://info",
    name="Server Information",
    description="Information about this MCP server"
)
async def server_info_resource() -> str:
    """Server information resource"""
    return f"""
MCP Server Information:
- Name: {server.name}
- Version: {server.version}
- Tools available: {len(server.tools)}
- Resources available: {len(server.resources)}
- Started at: {datetime.now().isoformat()}
- Environment: Hugging Face Spaces
"""


@server.resource(
    uri="server://capabilities",
    name="Server Capabilities",
    description="List of available tools and their descriptions"
)
async def capabilities_resource() -> str:
    """Server capabilities resource"""
    capabilities = "Available Tools:\n\n"
    for name, tool_info in server.tools.items():
        capabilities += f"- {name}: {tool_info['description']}\n"
    
    capabilities += "\nAvailable Resources:\n\n"
    for uri, resource_info in server.resources.items():
        capabilities += f"- {uri}: {resource_info['description']}\n"
    
    return capabilities


# Main execution function for stdio mode
async def main():
    """Main function for stdio-based MCP server"""
    server.logger.info("Starting MCP server in stdio mode")
    
    try:
        while True:
            # Read from stdin
            line = await asyncio.get_event_loop().run_in_executor(
                None, sys.stdin.readline
            )
            
            if not line:
                break
                
            try:
                request = json.loads(line.strip())
                response = await server.handle_request(request)
                print(json.dumps(response))
                sys.stdout.flush()
            except json.JSONDecodeError:
                error_response = {
                    "jsonrpc": "2.0",
                    "id": None,
                    "error": {
                        "code": -32700,
                        "message": "Parse error"
                    }
                }
                print(json.dumps(error_response))
                sys.stdout.flush()
                
    except KeyboardInterrupt:
        server.logger.info("Server shutting down")
    except Exception as e:
        server.logger.error(f"Server error: {e}")


if __name__ == "__main__":
    asyncio.run(main())
EOF

echo "📝 Creating http_wrapper.py..."
# Create http_wrapper.py with a truncated version due to length
cat > http_wrapper.py << 'EOF'
#!/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"""
    <!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
@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()
EOF

echo "📝 Creating requirements.txt..."
cat > requirements.txt << 'EOF'
fastapi>=0.104.1
uvicorn[standard]>=0.24.0
pydantic>=2.5.0
httpx>=0.25.0
requests>=2.31.0
python-json-logger>=2.0.7
python-multipart>=0.0.6
python-dotenv>=1.0.0
EOF

echo "📝 Creating Dockerfile..."
cat > Dockerfile << 'EOF'
FROM python:3.11-slim

WORKDIR /app

ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
ENV PORT=7860
ENV HOST=0.0.0.0

RUN apt-get update && apt-get install -y \
    build-essential \
    curl \
    git \
    && rm -rf /var/lib/apt/lists/* \
    && apt-get clean

COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

COPY . .

RUN useradd --create-home --shell /bin/bash app && \
    chown -R app:app /app
USER app

EXPOSE 7860

HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:7860/health || exit 1

CMD ["python", "http_wrapper.py"]
EOF

echo "📝 Creating .gitignore..."
cat > .gitignore << 'EOF'
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
htmlcov/
.tox/
.coverage
.pytest_cache/
*.log
.env
.venv
env/
venv/
.DS_Store
.vscode/
.idea/
Thumbs.db
*.tmp
*.swp
EOF

echo "📝 Creating README.md..."
cat > README.md << 'EOF'
---
title: HF MCP Server
emoji: 🤖
colorFrom: blue
colorTo: green
sdk: docker
pinned: false
license: mit
app_port: 7860
---

# MCP Server on Hugging Face Spaces

A Model Context Protocol (MCP) server implementation running on Hugging Face Spaces with Docker.

## 🚀 Features

- **Multiple Tools**: Echo, calculator, word counter, text reverser, current time
- **Resources**: Server information and capabilities  
- **HTTP API**: RESTful interface with automatic documentation
- **Docker Ready**: Optimized for Hugging Face Spaces deployment

## 🛠️ Available Tools

| Tool | Description | Parameters |
|------|-------------|------------|
| `echo` | Echo back input text | `text: string` |
| `current_time` | Get current timestamp | None |
| `calculate` | Basic math calculations | `expression: string` |
| `word_count` | Count words and characters | `text: string` |
| `reverse_text` | Reverse given text | `text: string` |

## 📚 API Endpoints

- `GET /` - Server information page
- `GET /health` - Health check
- `POST /mcp` - Main MCP protocol endpoint
- `GET /tools` - List all available tools
- `GET /docs` - Interactive API documentation

## 🔧 Usage Examples

### Health Check
```bash
curl https://your-space-name.hf.space/health
```

### List Tools
```bash
curl -X POST "https://your-space-name.hf.space/mcp" \
     -H "Content-Type: application/json" \
     -d '{"method": "tools/list"}'
```

### Call Echo Tool
```bash
curl -X POST "https://your-space-name.hf.space/mcp" \
     -H "Content-Type: application/json" \
     -d '{
       "method": "tools/call",
       "params": {
         "name": "echo",
         "arguments": {"text": "Hello World!"}
       }
     }'
```

## 🚀 Deployment to Hugging Face Spaces

1. Create a new Space on huggingface.co/spaces
2. Choose "Docker" as the SDK
3. Upload all files to your Space repository
4. The Space will automatically build and deploy

## 📄 License

MIT License - feel free to use and modify as needed.
EOF

echo "✅ Project setup complete!"
echo ""
echo "📂 Files created in: $PROJECT_DIR"
echo "📋 Next steps:"
echo "   1. cd $PROJECT_DIR"
echo "   2. Test locally: python http_wrapper.py"
echo "   3. Create Hugging Face Space and upload files"
echo "   4. Choose 'Docker' as SDK when creating the Space"
echo ""
echo "🌐 Local server will run on: http://localhost:7860"
echo "📚 API docs will be at: http://localhost:7860/docs"