API Bridge

从 OpenAPI/Swagger 规范生成 API 集成 —— 包括 MCP 服务器、API 客户端和 webhook 处理器。

何时使用

在用户满足以下任一条件时激活此技能：

• 说出“连接到这个 API”、“与 X 集成”、“构建 API 客户端”
• 想为 REST API 创建一个 MCP 服务器
• 拥有 OpenAPI/Swagger 规范并希望生成代码
• 需要处理 API 事件的 webhook 处理器
• 希望快速搭建 API 集成
• 提到“API 密钥”、“REST 端点”或“webhook”

此技能的功能

当用户希望连接到某个 API 时，该技能会：

1. 查找或获取 OpenAPI/Swagger 规范
2. 生成可运行的 MCP 服务器或 API 客户端
3. 处理认证（API 密钥、OAuth2、Bearer Token）
4. 包含错误处理、速率限制和重试机制
5. 生成可直接使用的代码

快速开始

从 OpenAPI 规范 URL 获取

# 获取规范
curl -s https://api.example.com/openapi.json > /tmp/api-spec.json

# 生成 MCP 服务器
# （请参考以下步骤）

从现有 API 文档创建

如果不存在 OpenAPI 规范，可基于 API 文档创建：

1. 确定基础 URL 和接口端点
2. 创建一个最小化的 OpenAPI 3.0 规范
3. 生成集成代码

第一步：解析规范

import json

with open('api-spec.json') as f:
    spec = json.load(f)

info = spec.get('info', {})
base_url = spec.get('servers', [{}])[0].get('url', '')
paths = spec.get('paths', {})

print(f"API: {info.get('title')}")
print(f"版本: {info.get('version')}")
print(f"基础地址: {base_url}")
print(f"端点数量: {len(paths)}")

# 列出所有端点
for path, methods in paths.items():
    for method in methods:
        if method in ['get', 'post', 'put', 'patch', 'delete']:
            print(f"  {method.upper()} {path}")

第二步：生成 MCP 服务器（Python）

# mcp_server.py - 由 OpenAPI 规范生成
import os
import json
import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent

app = Server("api-name")

BASE_URL = "https://api.example.com"
API_KEY = os.environ.get("API_KEY", "")

@app.tool()
async def list_items(limit: int = 10) -> str:
    """从 API 列出项目。"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/items",
            headers={"Authorization": f"Bearer {API_KEY}"},
            params={"limit": limit},
        )
        resp.raise_for_status()
        return json.dumps(resp.json(), indent=2)

@app.tool()
async def get_item(item_id: str) -> str:
    """根据 ID 获取单个项目。"""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{BASE_URL}/items/{item_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
        )
        resp.raise_for_status()
        return json.dumps(resp.json(), indent=2)

@app.tool()
async def create_item(name: str, description: str = "") -> str:
    """创建新项目。"""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{BASE_URL}/items",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={"name": name, "description": description},
        )
        resp.raise_for_status()
        return json.dumps(resp.json(), indent=2)

第三步：添加速率限制

import asyncio
from collections import deque
import time

class RateLimiter:
    def __init__(self, max_requests: int = 100, window: int = 60):
        self.max_requests = max_requests
        self.window = window
        self.timestamps = deque()

    async def acquire(self):
        now = time.time()
        while self.timestamps and now - self.timestamps[0] > self.window:
            self.timestamps.popleft()

        if len(self.timestamps) >= self.max_requests:
            wait = self.window - (now - self.timestamps[0])
            await asyncio.sleep(wait)

        self.timestamps.append(time.time())

rate_limiter = RateLimiter(max_requests=100, window=60)

第四步：添加重试逻辑

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
)
async def api_request(method: str, path: str, **kwargs) -> dict:
    """带有重试逻辑的 API 请求。"""
    async with httpx.AsyncClient() as client:
        resp = await client.request(
            method,
            f"{BASE_URL}{path}",
            headers={"Authorization": f"Bearer {API_KEY}"},
            **kwargs,
        )
        resp.raise_for_status()
        return resp.json()

生成 Python API 客户端

markdown

Api Bridge

API 客户端生成

"""基于 OpenAPI 规范生成的 API 客户端。"""

import httpx
from typing import Optional, Dict, Any

class APIClient:
    def __init__(self, base_url: str, api_key: str = ""):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self._client = httpx.AsyncClient(
            base_url=self.base_url,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json",
            },
            timeout=30.0,
        )

    async def close(self):
        await self._client.aclose()

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        await self.close()

    async def _request(
        self,
        method: str,
        path: str,
        params: Optional[Dict] = None,
        json: Optional[Dict] = None,
    ) -> Any:
        resp = await self._client.request(method, path, params=params, json=json)
        resp.raise_for_status()
        return resp.json()

    # 由 OpenAPI 规范生成的方法
    async def list_items(self, limit: int = 10) -> list:
        return await self._request("GET", "/items", params={"limit": limit})

    async def get_item(self, item_id: str) -> dict:
        return await self._request("GET", f"/items/{item_id}")

    async def create_item(self, name: str, **kwargs) -> dict:
        return await self._request("POST", "/items", json={"name": name, **kwargs})

    async def update_item(self, item_id: str, **kwargs) -> dict:
        return await self._request("PATCH", f"/items/{item_id}", json=kwargs)

    async def delete_item(self, item_id: str) -> None:
        await self._request("DELETE", f"/items/{item_id}")

Webhook 处理器生成

"""用于处理 API 事件的 Webhook 处理器。"""

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "")

@app.post("/webhook")
async def handle_webhook(request: Request):
    # 验证签名
    body = await request.body()
    signature = request.headers.get("X-Signature", "")

    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="无效签名")

    payload = await request.json()
    event_type = payload.get("event_type")

    if event_type == "item.created":
        await handle_item_created(payload)
    elif event_type == "item.updated":
        await handle_item_updated(payload)
    elif event_type == "item.deleted":
        await handle_item_deleted(payload)

    return {"status": "ok"}

常见 API 模式

OAuth2 认证流程

async def get_oauth_token(
    client_id: str,
    client_secret: str,
    token_url: str,
) -> str:
    """获取 OAuth2 访问令牌。"""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            token_url,
            data={
                "grant_type": "client_credentials",
                "client_id": client_id,
                "client_secret": client_secret,
            },
        )
        resp.raise_for_status()
        return resp.json()["access_token"]

分页处理

async def list_all_items(api_client) -> list:
    """自动分页列出所有项目。"""
    all_items = []
    cursor = None

    while True:
        params = {"limit": 100}
        if cursor:
            params["cursor"] = cursor

        result = await api_client._request("GET", "/items", params=params)
        all_items.extend(result["data"])

        cursor = result.get("next_cursor")
        if not cursor:
            break

    return all_items

文件上传

async def upload_file(api_client, file_path: str) -> dict:
    """将文件上传至 API。"""
    async with httpx.AsyncClient() as client:
        with open(file_path, "rb") as f:
            resp = await client.post(
                f"{api_client.base_url}/files",
                headers={"Authorization": f"Bearer {api_client.api_key}"},
                files={"file": (file_path, f, "application/octet-stream")},
            )
            resp.raise_for_status()
            return resp.json()

OpenAPI 规范来源

常见可查找 OpenAPI 规范的位置：

也可通过搜索：site:github.com "openapi" OR "swagger" "api-name" 查找。

常见陷阱

• 规范版本 — 确认是 OpenAPI 2.0（Swagger）还是 3.0+（格式不同）
• 认证方式 — 区分 API Key、Bearer Token、OAuth2、Basic Auth（查看 spec 中的 securitySchemes）
• 速率限制 — 始终实现限流；查阅 API 文档了解具体限制
• 分页机制 — 不同 API 使用不同分页方式（游标、偏移量、页码、Link 头）
• 内容类型 — 部分 API 使用 application/x-www-form-urlencoded 而非 JSON
• Webhook 验证 — 必须验证签名；不要信任原始负载数据
• 错误处理 — 查阅 API 文档中的错误码和 retry-after 头信息

验证步骤

生成集成后，请按以下步骤验证：

1. 测试简单 GET 请求（如列表接口）
2. 验证认证是否生效（检查响应状态码）
3. 测试错误处理（如无效 ID、缺少认证）
4. 检查速率限制（超过限制，验证退避行为）
5. 测试分页功能（如适用）