从 OpenAPI 规范生成 MCP 服务器可将任何记录的 REST API 转换为 AI 代理可以直接调用的一组类型化工具，从而消除手动工具定义。您无需为每个端点手动编写 MCP 工具处理程序，而是解析 OpenAPI JSON/YAML，提取操作、参数和响应架构，并生成 Claude Code、Cursor 或任何 MCP 客户端可以使用的符合标准的 MCP 服务器。本教程使用 Python 构建一个生成器，该生成器采用 OpenAPI 3.x 规范并输出工作的 MCP 服务器。
前置条件
- Python 3.10+
- 安装 fastmcp 库 (pip install fastmcp)
- OpenAPI 3.x 规范（JSON 或 YAML）
- 用于使用真实规格进行测试的 Scadio API 密钥
操作指南
步骤 1: 解析 OpenAPI 规范
加载并解析 OpenAPI 3.x 规范文件，提取所有操作及其路径、方法、参数和请求正文。
Python
import json, yaml
from pathlib import Path
def parse_openapi(spec_path):
content = Path(spec_path).read_text()
spec = yaml.safe_load(content) if spec_path.endswith(('.yml', '.yaml')) else json.loads(content)
operations = []
for path, methods in spec.get('paths', {}).items():
for method, details in methods.items():
if method in ('get', 'post', 'put', 'patch', 'delete'):
op = {
'operation_id': details.get('operationId', f'{method}_{path}'.replace('/', '_')),
'summary': details.get('summary', ''),
'method': method.upper(),
'path': path,
'parameters': details.get('parameters', []),
'request_body': details.get('requestBody', {}),
}
operations.append(op)
print(f'Parsed {len(operations)} operations from {spec_path}')
return spec, operations
spec, ops = parse_openapi('openapi.json')步骤 2: 生成 MCP 工具定义
将每个 OpenAPI 操作转换为具有类型化参数和描述的 MCP 工具。
Python
def openapi_type_to_python(schema):
type_map = {'string': 'str', 'integer': 'int', 'number': 'float',
'boolean': 'bool', 'array': 'list', 'object': 'dict'}
return type_map.get(schema.get('type', 'string'), 'str')
def generate_tool_code(op, base_url):
params = []
for p in op['parameters']:
ptype = openapi_type_to_python(p.get('schema', {}))
default = f' = None' if not p.get('required') else ''
params.append(f"{p['name']}: {ptype}{default}")
body_params = []
if op['request_body']:
content = op['request_body'].get('content', {})
json_schema = content.get('application/json', {}).get('schema', {})
for prop, schema in json_schema.get('properties', {}).items():
ptype = openapi_type_to_python(schema)
required = prop in json_schema.get('required', [])
default = '' if required else ' = None'
body_params.append(f"{prop}: {ptype}{default}")
all_params = ', '.join(params + body_params)
name = op['operation_id'].replace('-', '_').replace('.', '_')
return name, all_params, op['summary']步骤 3: 构建 MCP 服务器文件
生成完整的 MCP 服务器 Python 文件，用于注册所有工具并处理对底层 API 的 HTTP 请求。
Python
def generate_mcp_server(spec, operations, base_url, output_path='generated_mcp.py'):
lines = [
'import httpx',
'from fastmcp import FastMCP',
'',
f'mcp = FastMCP("Generated API")',
f'BASE_URL = "{base_url}"',
'',
]
for op in operations:
name, params, summary = generate_tool_code(op, base_url)
lines.append(f'@mcp.tool(description="{summary}")')
lines.append(f'async def {name}({params}) -> dict:')
lines.append(f' async with httpx.AsyncClient() as client:')
if op['method'] == 'GET':
lines.append(f' resp = await client.get(f"{{BASE_URL}}{op["path"]}")')
else:
lines.append(f' resp = await client.{op["method"].lower()}(f"{{BASE_URL}}{op["path"]}", json=locals())')
lines.append(f' return resp.json()')
lines.append('')
Path(output_path).write_text('\n'.join(lines))
print(f'Generated MCP server with {len(operations)} tools at {output_path}')
generate_mcp_server(spec, ops, 'https://api.example.com')步骤 4: 使用 Scavio OpenAPI 规范进行测试
使用搜索获取 Scavio API 规范，并从中生成 MCP 服务器作为工作示例。
Python
import requests, os
# Use Scavio search to find OpenAPI specs for testing
H = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'query': 'petstore openapi 3.0 json spec', 'country_code': 'us'}).json()
print('Found specs to test with:')
for r in data.get('organic_results', [])[:3]:
print(f" {r.get('title', '')}: {r.get('link', '')}")
# Example: generate from the classic Petstore spec
# spec, ops = parse_openapi('petstore.json')
# generate_mcp_server(spec, ops, 'https://petstore3.swagger.io/api/v3')Python 示例
Python
import json, yaml, httpx, os, requests
from pathlib import Path
def parse_openapi(spec_path):
content = Path(spec_path).read_text()
spec = yaml.safe_load(content) if spec_path.endswith(('.yml', '.yaml')) else json.loads(content)
ops = []
for path, methods in spec.get('paths', {}).items():
for method, details in methods.items():
if method in ('get', 'post', 'put', 'patch', 'delete'):
ops.append({
'id': details.get('operationId', f'{method}_{path}'.replace('/', '_')),
'summary': details.get('summary', ''),
'method': method.upper(), 'path': path,
'parameters': details.get('parameters', []),
'body': details.get('requestBody', {}),
})
return spec, ops
def generate_server(spec, ops, base_url, out='generated_mcp.py'):
type_map = {'string': 'str', 'integer': 'int', 'number': 'float', 'boolean': 'bool'}
lines = ['import httpx', 'from fastmcp import FastMCP', '',
'mcp = FastMCP("Generated API")', f'BASE = "{base_url}"', '']
for op in ops:
name = op['id'].replace('-', '_').replace('.', '_')
params = []
for p in op['parameters']:
pt = type_map.get(p.get('schema', {}).get('type', 'string'), 'str')
params.append(f"{p['name']}: {pt}")
sig = ', '.join(params)
lines.extend([
f'@mcp.tool(description="{op["summary"]}")',
f'async def {name}({sig}) -> dict:',
f' async with httpx.AsyncClient() as c:',
f' r = await c.request("{op["method"]}", f"{{BASE}}{op["path"]}")',
f' return r.json()', ''
])
Path(out).write_text('\n'.join(lines))
print(f'Generated {len(ops)} MCP tools to {out}')
# Example usage
# spec, ops = parse_openapi('openapi.json')
# generate_server(spec, ops, 'https://api.example.com')
print('MCP generator ready. Pass any OpenAPI 3.x spec.')JavaScript 示例
JavaScript
const fs = require('fs');
function parseOpenAPI(specPath) {
const spec = JSON.parse(fs.readFileSync(specPath, 'utf-8'));
const ops = [];
for (const [path, methods] of Object.entries(spec.paths || {})) {
for (const [method, details] of Object.entries(methods)) {
if (['get', 'post', 'put', 'patch', 'delete'].includes(method)) {
ops.push({
id: details.operationId || \`\${method}_\${path.replace(/\//g, '_')}\`,
summary: details.summary || '',
method: method.toUpperCase(), path,
parameters: details.parameters || [],
});
}
}
}
return {spec, ops};
}
function generateServer(spec, ops, baseUrl, out = 'generated_mcp.mjs') {
let code = \`import {FastMCP} from 'fastmcp';\nconst mcp = new FastMCP('Generated API');\nconst BASE = '\${baseUrl}';\n\n\`;
for (const op of ops) {
const name = op.id.replace(/[-. ]/g, '_');
const params = op.parameters.map(p => \`\${p.name}: {type: 'string'}\`).join(', ');
code += \`mcp.addTool({name: '\${name}', description: '\${op.summary}',\n parameters: {\${params}},\n execute: async (args) => {\n const r = await fetch(\\\`\\\${BASE}\${op.path}\\\`, {method: '\${op.method}'});\n return JSON.stringify(await r.json());\n }\n});\n\n\`;
}
fs.writeFileSync(out, code);
console.log(\`Generated \${ops.length} MCP tools to \${out}\`);
}
// const {spec, ops} = parseOpenAPI('openapi.json');
// generateServer(spec, ops, 'https://api.example.com');
console.log('MCP generator ready');预期输出
JSON
Parsed 12 operations from openapi.json
Generated MCP server with 12 tools at generated_mcp.py
Tools: get_users, post_users, get_users_by_id, put_users_by_id, ...