我如何用 50 行 Python 构建 MCP 服务器（自动生成自 OpenAPI）

Source: Dev.to

我们要构建的内容

一个 MCP 服务器，具备以下功能：

• ✅ 自动从任意 OpenAPI 规范生成工具
• ✅ 处理每个请求的身份验证（API 密钥）
• ✅ 与 Claude Desktop、Cursor 以及任何 MCP 客户端兼容
• ✅ 以标准 Web 服务形式部署，并提供健康检查

结果： AI 助手可以通过自然语言执行 LinkedIn 操作（发布评论、获取个人资料、管理互动）。

技术栈

架构

┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│  Claude Desktop │────▶│   MCP 服务器    │────▶│  LinkedIn API   │
│  / Cursor / AI  │     │  (FastMCP)      │     │  (OpenAPI)      │
└─────────────────┘     └─────────────────┘     └─────────────────┘
        │                       │
        │   ?apiKey=xxx         │  Authorization: Bearer xxx
        └───────────────────────┘
              API 密钥流程

中间层读取 OpenAPI 规范，并自动将每个端点公开为 MCP 工具。

完整代码

1. 设置每请求身份验证

from contextvars import ContextVar
import httpx

# Store API key per request using context variables (thread‑safe)
api_key_context: ContextVar[str] = ContextVar('api_key', default='')

# Event hook to inject Authorization header from context
async def add_auth_header(request: httpx.Request):
    """
    Async event hook that injects the Authorization header 
    from the context variable. Called for every outgoing request.
    """
    api_key = api_key_context.get()
    if api_key:
        request.headers["Authorization"] = f"Bearer {api_key}"

Why ContextVar? Unlike globals, ContextVar maintains separate values for each async context, preventing key leakage between concurrent requests.

2. 使用事件钩子创建 HTTP 客户端

api_client = httpx.AsyncClient(
    base_url="https://api.connectsafely.ai/linkedin",
    timeout=300.0,  # 5‑minute timeout for long operations
    event_hooks={"request": [add_auth_header]}
)

The event_hooks pattern ensures every request automatically receives the correct Authorization header.

3. 加载 OpenAPI 规范

import sys
import httpx

try:
    print("Loading OpenAPI spec...")
    openapi_spec = httpx.get(
        "https://api.connectsafely.ai/linkedin/openapi.json",
        timeout=5.0
    ).json()
    print("OpenAPI spec loaded successfully!")
except httpx.ConnectError:
    print("ERROR: Could not connect to API")
    sys.exit(1)
except Exception as e:
    print(f"ERROR: Failed to load OpenAPI spec: {e}")
    sys.exit(1)

Benefits:

• Zero manual tool definitions – endpoints become tools automatically
• Always in sync – updating the API updates MCP tools automatically
• Self‑documenting – tool descriptions come from OpenAPI

4. 魔法：从 OpenAPI 创建 FastMCP

from fastmcp import FastMCP

# Create the MCP server from the OpenAPI specification
mcp = FastMCP.from_openapi(
    openapi_spec=openapi_spec,
    client=api_client,
    name="LinkedIn API Server"
)

A single line converts the entire OpenAPI spec into MCP tools, each with proper parameter validation, type hints, and documentation.

5. FastAPI 集成以实现生产特性

from fastapi import FastAPI, Request

# Create MCP ASGI app
mcp_app = mcp.http_app(path='/')

# Create FastAPI app with MCP lifespan
app = FastAPI(lifespan=mcp_app.lifespan)

@app.get("/health")
async def health_check():
    """Health check endpoint for Docker and load balancers."""
    return {"status": "healthy"}

FastAPI adds health checks, custom middleware, and additional REST endpoints as needed.

6. API 密钥提取中间件

@app.middleware("http")
async def extract_api_key(request: Request, call_next):
    """
    Middleware to extract API key from query parameter 
    or Authorization header. Supports both formats:
    - ?apiKey=your-key
    - Authorization: Bearer your-key
    """
    # Extract from query parameter first
    api_key = request.query_params.get("apiKey", "")

    if not api_key:
        # Fallback to Authorization header
        auth_header = request.headers.get("Authorization", "")
        if auth_header.startswith("Bearer "):
            api_key = auth_header.split(" ")[1]

    # Store in context for downstream use
    api_key_context.set(api_key)

    response = await call_next(request)
    return response

# Mount MCP server at root
app.mount("/", mcp_app)

The middleware captures the API key for each request and stores it in the ContextVar, which the httpx hook later reads.

7. 运行服务器

if __name__ == "__main__":
    import uvicorn

    print("\nStarting MCP server on http://0.0.0.0:3011")
    print("Connect with: http://localhost:3011?apiKey=your-api-key")

    uvicorn.run(app, host="0.0.0.0", port=3011)

8. 连接到 Claude Desktop

Add the following to claude_desktop_config.json:

{
  "mcpServers": {
    "connectsafely": {
      "url": "http://localhost:3011?apiKey=YOUR_API_KEY"
    }
  }
}

Claude can now execute LinkedIn automation tasks via natural language, e.g.:

“在最新的帖子上给 @naval 留言，内容是 ‘Great insight on leverage!’”

Claude will:

1. Search for the user’s posts
2. Find the latest one
3. Post the comment through the MCP‑generated tool

为什么这种模式强大

1. 零重复

Your REST API and MCP tools share the same OpenAPI source of truth. Updating the spec updates both automatically.