🔌 MCP服务器开发详解

# 使用Python实现一个简单的MCP Server
import json
from http.server import HTTPServer, BaseHTTPRequestHandler

class MCPServer(BaseHTTPRequestHandler):
    # 注册可用工具
    tools = {
        "search_web": {
            "description": "搜索引擎查询，返回网页标题、摘要和链接",
            "parameters": {
                "query": {"type": "string", "description": "搜索关键词"},
                "limit": {"type": "integer", "description": "返回结果数，默认5"}
            }
        },
        "get_current_time": {
            "description": "获取指定时区的当前时间",
            "parameters": {
                "timezone": {"type": "string", "description": "时区，如'Asia/Shanghai'"}
            }
        }
    }
    
    def do_POST(self):
        content_length = int(self.headers['Content-Length'])
        body = json.loads(self.rfile.read(content_length))
        response = self.handle_mcp_request(body)
        
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps(response).encode())
    
    def handle_mcp_request(self, request):
        method = request["method"]
        
        if method == "tools/list":
            return self.list_tools(request)
        elif method == "tools/call":
            return self.call_tool(request)
        elif method == "resources/list":
            return self.list_resources(request)
        else:
            return {
                "jsonrpc": "2.0",
                "error": {"code": -32601, "message": "方法不存在"},
                "id": request.get("id")
            }

# 启动服务器
if __name__ == "__main__":
    server = HTTPServer(("localhost", 8080), MCPServer)
    print(f"MCP Server 已启动，监听 0.0.0.0:8080")
    print(f"可用工具: {list({'search_web', 'get_current_time'})}")
    server.serve_forever()

def call_tool(self, request):
    tool_name = request["params"]["name"]
    args = request["params"].get("arguments", {})
    
    tool_map = {
        "search_web": self.search_web,
        "get_current_time": self.get_current_time
    }
    
    if tool_name not in tool_map:
        return {
            "jsonrpc": "2.0",
            "error": {"code": -32602, "message": f"工具 '{tool_name}' 不存在"},
            "id": request["id"]
        }
    
    result = tool_map[tool_name](args)
    return {
        "jsonrpc": "2.0",
        "result": result,
        "id": request["id"]
    }

def get_current_time(self, args):
    import pytz
    from datetime import datetime
    
    tz = pytz.timezone(args.get("timezone", "Asia/Shanghai"))
    now = datetime.now(tz)
    return {
        "content": [{"type": "text", "text": now.strftime("%Y-%m-%d %H:%M:%S %Z")}]
    }

# API Key认证中间件
class AuthMiddleware:
    def __init__(self):
        # 存储有效的API Keys
        self.api_keys = {
            "sk-miaoquai-...": {"permissions": ["read", "write"]},
            "sk-readonly-...": {"permissions": ["read"]}
        }
    
    def authenticate(self, headers):
        auth_header = headers.get("Authorization", "")
        api_key = auth_header.replace("Bearer ", "")
        
        if api_key not in self.api_keys:
            return None  # 认证失败
        
        return self.api_keys[api_key]  # 返回权限信息
    
    def check_permission(self, tool_name, permissions):
        # 检查是否有权限调用该工具
        if tool_name in ["delete_database", "modify_config"]:
            return "write" in permissions
        return True  # 只读工具允许所有认证用户访问