OpenAPI-to-MCP Bridge: 一键将 Spring Boot OpenAPI 转换为 MCP 工具服务器

📝 项目介绍 (Project Description)

简介

OpenAPI-to-MCP Bridge 是一个轻量级的 Python Web 服务器，旨在打破传统后端服务与大语言模型（LLM）Agent 之间的壁垒。它能够自动拉取现有的 Spring Boot 应用生成的 OpenAPI (Swagger) 规范，实时解析接口定义、参数约束及数据结构，并将其动态注册为符合 Model Context Protocol (MCP) 标准的工具（Tools）。

通过本项目，你的 LLM Agent 无需任何手动配置，即可“理解”并调用现有的 RESTful API，实现从自然语言指令到后端业务逻辑的自动执行。

✨ 核心特性

• 🔄 零代码适配 (Zero-Code Integration) 只需提供 Spring Boot 应用的 base-url，脚本自动获取 /v3/api-docs，无需编写任何胶水代码即可暴露所有 GET/POST/PUT/DELETE 接口。
• 🧠 智能 Schema 解析 (Smart Schema Resolution) 内置 prance 解析器，自动处理 OpenAPI 中的 $ref 引用循环和嵌套结构，确保复杂对象也能被准确识别。
• 📝 自动生成高质量示例 (Auto-Generated Examples) 独创的 schema_to_example 算法，根据字段类型、格式（如 date-time, uuid）、枚举值及默认值，动态生成逼真的 JSON 请求体示例。这不仅帮助开发者理解接口，更显著提升了 LLM 构造正确参数的准确率。
• 🛠️ 动态工具构建 (Dynamic Tool Generation) 利用 Python 元编程技术，运行时动态构建异步 HTTP 客户端函数。自动区分 Path 参数、Query 参数和 Request Body，并生成包含详细文档（Docstring）的工具描述。
• ⚡ 基于 FastMCP 标准 完全兼容 MCP 协议，支持 SSE (Server-Sent Events) 传输模式，可无缝对接 Cursor、Claude Desktop 或其他支持 MCP 的 AI 客户端。

🌟 工作原理

🛠️ 快速开始

确保已安装依赖：

pip install fastmcp httpx prance openapi-spec-validator

运行服务器：

python main.py –base-url http://localhost:8080

程序将自动加载 http://localhost:8080/v3/api-docs 并在 0.0.0.0:8000 启动 MCP 服务。

💡 应用场景

• 遗留系统现代化: 快速让旧有的 Spring Boot 微服务具备 AI Agent 交互能力。
• 内部助手开发: 为企业内部知识库或运维助手自动挂载数据库管理、订单查询等 API 工具。
• 原型验证: 在 API 设计阶段，立即测试 LLM 对接口理解的准确性。

技术栈

• Core: Python 3.12+
• MCP Framework: FastMCP
• HTTP Client: httpx (Async)
• Spec Parser: prance + openapi-spec-validator
• Target: Spring Boot (OpenAPI 3.0)

代码

# 导入所需的库
import json
import httpx
from fastmcp import FastMCP
import prance # 用于解析和 dereference OpenAPI spec

# — 使用 prance 自动解析 OpenAPI spec —
def load_and_resolve_openapi_spec(source: str) -> dict:
"""
从 URL 或本地文件加载 OpenAPI spec，并自动解析所有 $ref 引用（dereference）。

返回一个完全展开的、无 $ref 的规范字典。
"""
parser = prance.ResolvingParser(
source,
strict=False, # 宽松模式，容忍部分非标准字段
backend="openapi-spec-validator"
)
return parser.specification

# — 简化后的 schema_to_example（不再需要 resolve 逻辑）—
def schema_to_example(schema: dict, depth: int = 0, max_depth: int = 4):
if depth > max_depth or not isinstance(schema, dict):
return None

# OpenAPI example/enum/default 优先
if "example" in schema:
return schema["example"]
if "default" in schema:
return schema["default"]
if "enum" in schema and schema["enum"]:
return schema["enum"][0]

# 处理 oneOf / anyOf：取第一个（为了示例稳定）
for key in ("oneOf", "anyOf"):
if key in schema and schema[key]:
return schema_to_example(schema[key][0], depth, max_depth)

t = schema.get("type")
fmt = schema.get("format")

if t == "string" or (t is None and "properties" not in schema and "items" not in schema):
if fmt in ("date-time", "datetime"):
return "2026-02-06T12:00:00Z"
if fmt == "date":
return "2026-02-06"
if fmt == "uuid":
return "00000000-0000-0000-0000-000000000000"
return "string"

if t == "integer":
return 0
if t == "number":
return 0.0
if t == "boolean":
return True

if t == "array":
item_schema = schema.get("items", {})
return [schema_to_example(item_schema, depth + 1, max_depth)]

# object 类型
if t == "object" or "properties" in schema or "additionalProperties" in schema:
obj = {}
props = schema.get("properties") or {}
required = set(schema.get("required", []))

# 先填 required 字段
for name in required:
if name in props:
obj[name] = schema_to_example(props[name], depth + 1, max_depth)

# 再补 1~2 个 optional 字段（避免示例太简单）
optional_count = 0
for name, ps in props.items():
if name in obj:
continue
obj[name] = schema_to_example(ps, depth + 1, max_depth)
optional_count += 1
if optional_count >= 2:
break

# additionalProperties 支持
addl = schema.get("additionalProperties")
if isinstance(addl, dict):
obj.setdefault("exampleKey", schema_to_example(addl, depth + 1, max_depth))

return obj

return None

def build_request_body_example(spec: dict, details: dict):
rb = details.get("requestBody") or {}
content = rb.get("content") or {}
app_json = content.get("application/json") or {}
schema = app_json.get("schema")
if not schema:
return None
return schema_to_example(schema)

# 创建 FastMCP 实例
mcp = FastMCP(name="OpenAPI Bridge", instructions="Expose OpenAPI endpoints as MCP tools")

def register_openapi_tools(spec: dict, base_url: str):
paths = spec.get("paths", {})

for path, methods in paths.items():
for method, details in methods.items():
if method not in ["get", "post", "put", "delete"]:
continue

operation_id = details.get("operationId", f"{method}_{path.replace('/', '_').strip('_')}")
description = details.get("summary", details.get("description", ""))

# 提取 path/query 参数
parameters = details.get("parameters", [])
param_defs = []
for p in parameters:
if p.get("in") not in ("path", "query"):
continue
name = p.get("name")
if not name:
continue
required = bool(p.get("required", False)) or p.get("in") == "path"
param_defs.append((name, required))

# 去重（保留首次出现）
seen = set()
unique_param_defs = []
for n, r in param_defs:
if n not in seen:
seen.add(n)
unique_param_defs.append((n, r))
param_defs = unique_param_defs

# requestBody 示例
body_example = build_request_body_example(spec, details)

# 构建 docstring
doc_parts = [f"{description}\\n"]
if body_example is not None:
doc_parts.append("\\n 示例入参(JSON body):")
doc_parts.append(
" " + json.dumps(body_example, ensure_ascii=False, indent=2).replace("\\n", "\\n "))

responses = details.get("responses", {})
success_resp = responses.get("200") or responses.get("201") or {}
if success_resp:
desc = success_resp.get("description", "API 调用的响应文本。")
doc_parts.append(f"\\n Returns:\\n str: {desc}")

dynamic_docstring = "".join(doc_parts)

# 动态生成工具函数
def make_tool(p=path, m=method, url=base_url, _param_defs=None, _has_body=False):
_param_defs = _param_defs or []

lines = ["async def tool_func("]
for name, required in _param_defs:
if required:
lines.append(f" {name},")
else:
lines.append(f" {name}=None,")
if _has_body:
lines.append(" body=None,\\n")
lines.append(") -> str:\\n")
lines.append(" import httpx\\n")
lines.append(" async with httpx.AsyncClient() as client:\\n")
lines.append(f" full_url = f\\"{url}{p}\\"\\n")

# 替换路径参数
for name, _ in _param_defs:
placeholder = f"{{{name}}}"
if placeholder in p:
lines.append(f" full_url = full_url.replace('{placeholder}', str({name}))\\n")

# 构建 query 参数
lines.append(" params = {}\\n")
for name, _ in _param_defs:
placeholder = f"{{{name}}}"
if placeholder not in p: # 不是 path 参数 → 是 query
lines.append(f" if {name} is not None:\\n")
lines.append(f" params['{name}'] = {name}\\n")

# 发送请求
lines.append(f" method = '{m}'.upper()\\n")
if m.lower() == "get":
lines.append(" resp = await client.get(full_url, params=params)\\n")
else:
if _has_body:
lines.append(
" resp = await client.request(method, full_url, params=params, json=body)\\n")
else:
lines.append(" resp = await client.request(method, full_url, params=params)\\n")
lines.append(" return resp.text\\n")

ns = {}
print("—————-")
print("".join(lines))
print("—————-")
exec("".join(lines), ns, ns)
return ns["tool_func"]

tool = make_tool(_param_defs=param_defs, _has_body=(body_example is not None))
tool.__name__ = operation_id
tool.__doc__ = dynamic_docstring
mcp.tool(name=operation_id, description=dynamic_docstring)(tool)

import argparse

if __name__ == "__main__":
parser = argparse.ArgumentParser(description="启动 MCP 服务器并注册 OpenAPI 工具")
parser.add_argument(
"–base-url",
type=str,
required=True,
help="API 的基础 URL（例如 http://127.0.0.1:8080）"
)
args = parser.parse_args()

openapi_url = f"{args.base_url.rstrip('/')}/v3/api-docs"

print("Loading and resolving OpenAPI spec…")
spec = load_and_resolve_openapi_spec(openapi_url)

print("Registering tools…")
register_openapi_tools(spec, args.base_url)

print("Starting MCP server on http://0.0.0.0:8000 (SSE transport)…")
mcp.run(host="0.0.0.0", port=8000, transport="sse")