如何根据 Swagger 规范构建 MCP 服务器

import json, yaml
from pathlib import Path

def parse_openapi(spec_path: str) -> list:
    """Parse OpenAPI spec into MCP tool definitions."""
    raw = Path(spec_path).read_text()
    spec = yaml.safe_load(raw) if spec_path.endswith('.yaml') else json.loads(raw)
    tools = []
    base_url = spec.get('servers', [{}])[0].get('url', '')
    for path, methods in spec.get('paths', {}).items():
        for method, details in methods.items():
            if method not in ('get', 'post', 'put', 'delete'):
                continue
            params = []
            for p in details.get('parameters', []):
                params.append({
                    'name': p['name'],
                    'type': p.get('schema', {}).get('type', 'string'),
                    'required': p.get('required', False),
                    'description': p.get('description', '')
                })
            # Extract request body schema
            body = details.get('requestBody', {}).get('content', {}).get('application/json', {})
            body_schema = body.get('schema', {}).get('properties', {})
            for name, prop in body_schema.items():
                params.append({
                    'name': name, 'type': prop.get('type', 'string'),
                    'required': name in body.get('schema', {}).get('required', []),
                    'description': prop.get('description', '')
                })
            tool_name = details.get('operationId', f'{method}_{path}'.replace('/', '_'))
            tools.append({
                'name': tool_name, 'description': details.get('summary', ''),
                'method': method.upper(), 'path': path,
                'base_url': base_url, 'parameters': params
            })
    print(f'Parsed {len(tools)} tools from spec')
    for t in tools[:5]:
        print(f'  {t["name"]}: {t["method"]} {t["path"]}')
    return tools

def generate_mcp_server(tools: list, output_path: str = 'mcp_server.py'):
    """Generate a Python MCP server from parsed tool definitions."""
    lines = [
        'import json, os, sys, requests',
        'from typing import Any',
        '',
        'def handle_tool_call(name: str, arguments: dict) -> dict:',
        '    """Route tool calls to the correct API endpoint."""',
        '    tools_map = {',
    ]
    for tool in tools:
        lines.append(f'        "{tool["name"]}": {{"method": "{tool["method"]}", "path": "{tool["path"]}", "base_url": "{tool["base_url"]}"}},')  
    lines.extend([
        '    }',
        '    if name not in tools_map:',
        '        return {"error": f"Unknown tool: {name}"}',  
        '    spec = tools_map[name]',
        '    url = spec["base_url"] + spec["path"]',
        '    headers = {"Content-Type": "application/json",',
        '               "x-api-key": os.environ.get("API_KEY", "")}',
        '    if spec["method"] == "GET":',
        '        resp = requests.get(url, headers=headers, params=arguments)',
        '    else:',
        '        resp = requests.post(url, headers=headers, json=arguments)',
        '    return resp.json()',
        '',
    ])
    # Write MCP stdio loop
    lines.extend([
        'def main():',
        '    for line in sys.stdin:',
        '        msg = json.loads(line.strip())',
        '        if msg.get("method") == "tools/call":',
        '            result = handle_tool_call(msg["params"]["name"], msg["params"].get("arguments", {}))',
        '            print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"content": [{"type": "text", "text": json.dumps(result)}]}}))',
        '            sys.stdout.flush()',
        '',
        'if __name__ == "__main__":',
        '    main()',
    ])
    with open(output_path, 'w') as f:
        f.write('\n'.join(lines))
    print(f'Generated MCP server: {output_path}')
    print(f'Tools: {len(tools)}')

import json
from pathlib import Path

def register_mcp(server_name: str, server_path: str):
    mcp_config_path = Path('.mcp.json')
    config = json.loads(mcp_config_path.read_text()) if mcp_config_path.exists() else {'mcpServers': {}}
    config['mcpServers'][server_name] = {
        'command': 'python3',
        'args': [server_path],
        'env': {
            'API_KEY': 'your-api-key-here'
        }
    }
    mcp_config_path.write_text(json.dumps(config, indent=2))
    print(f'Registered MCP server: {server_name}')
    print(f'Config: {mcp_config_path}')
    print(f'Restart Claude Code to load the new tools.')

# Register the generated server
register_mcp('my-api', 'mcp_server.py')