创建自己的MCP Server

mkdir mcp-demo
cd mcp-demo
touch server.py

from typing import Any
from mcp.server.fastmcp import FastMCP
import subprocess
import shlex
import os

# Initialize FastMCP server
mcp = FastMCP("ShellExecutor")

# List of allowed commands (for security)
ALLOWED_COMMANDS = {
    'ls': True,
    'pwd': True,
    'echo': True,
    'cat': True,
    'grep': True,
    'find': True,
    'date': True,
    'whoami': True,
    'uname': True,
    'df': True,
    'du': True,
    'ps': True,
    'top': True,
    'free': True,
    'uptime': True,
    'netstat': True,
    'ifconfig': True,
    'ping': True,
    'traceroute': True,
    'dig': True,
    'nslookup': True,
    'curl': True,
    'wget': True,
    'git': True,
    'docker': True,
    'kubectl': True,
}

def is_command_allowed(command: str) -> bool:
    """Check if the command is in the allowed list."""
    cmd = shlex.split(command)[0]
    return cmd in ALLOWED_COMMANDS

@mcp.tool()
async def execute_command(command: str) -> dict[str, Any]:
    """
    Execute a shell command and return the result.
    
    Args:
        command: The shell command to execute
        
    Returns:
        A dictionary containing:
        - success: bool indicating if the command executed successfully
        - output: str containing the command output
        - error: str containing any error message
    """
    if not is_command_allowed(command):
        return {
            "success": False,
            "output": "",
            "error": f"Command '{command.split()[0]}' is not allowed"
        }
    
    try:
        # Execute the command with a timeout of 30 seconds
        result = subprocess.run(
            command,
            shell=True,
            capture_output=True,
            text=True,
            timeout=30
        )
        
        return {
            "success": True,
            "output": result.stdout,
            "error": result.stderr
        }
    except subprocess.TimeoutExpired:
        return {
            "success": False,
            "output": "",
            "error": "Command execution timed out after 30 seconds"
        }
    except Exception as e:
        return {
            "success": False,
            "output": "",
            "error": str(e)
        }

if __name__ == "__main__":
    # Run the MCP server
    mcp.run(transport='stdio')