REST API MCP 封装模式

将第三方 REST API 封装成 MCP Server，是扩展 AI 工具能力的最灵活方式。本节讲解三种封装模式，以及如何设计对 LLM 友好的工具接口。

三种封装模式

模式一：透传封装（快速起步）

直接把 API 端点包装成工具，适合快速验证：

# api_passthrough_server.py
import httpx
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
app = Server("shopify-api")
BASE_URL = "https://mystore.myshopify.com/admin/api/2024-01"
HEADERS = {"X-Shopify-Access-Token": "shpat_xxx", "Content-Type": "application/json"}
@app.list_tools()
async def list_tools():
return [
types.Tool(
name="get_products",
description="获取 Shopify 商品列表",
inputSchema={
"type": "object",
"properties": {
"limit": {"type": "integer", "default": 10, "maximum": 250},
"status": {"type": "string", "enum": ["active", "draft", "archived"]},
"title": {"type": "string", "description": "按标题筛选"}
}
}
),
types.Tool(
name="get_orders",
description="获取订单列表",
inputSchema={
"type": "object",
"properties": {
"status": {"type": "string", "enum": ["open", "closed", "cancelled", "any"]},
"limit": {"type": "integer", "default": 10},
"created_at_min": {"type": "string", "description": "ISO 日期，如 2026-03-01"}
}
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
async with httpx.AsyncClient() as client:
if name == "get_products":
response = await client.get(
f"{BASE_URL}/products.json",
headers=HEADERS,
params={k: v for k, v in arguments.items() if v is not None}
)
elif name == "get_orders":
response = await client.get(
f"{BASE_URL}/orders.json",
headers=HEADERS,
params=arguments
)
else:
raise ValueError(f"未知工具: {name}")
data = response.json()
return [types.TextContent(type="text", text=json.dumps(data, ensure_ascii=False))]

模式二：聚合封装（减少工具调用）

将多个 API 调用合并为一个高层工具，减少 LLM 的推理步骤：

# 聚合工具：获取订单的完整信息（订单+客户+商品）
@app.list_tools()
async def list_tools():
return [
types.Tool(
name="get_order_full_details",
description="获取订单的完整信息，包括客户信息、订单行、履约状态（一次调用获取所有相关数据）",
inputSchema={
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "订单ID"}
},
"required": ["order_id"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
order_id = arguments["order_id"]
async with httpx.AsyncClient() as client:
# 并发请求3个端点
order_task = client.get(f"{BASE_URL}/orders/{order_id}.json", headers=HEADERS)
fulfillment_task = client.get(
f"{BASE_URL}/orders/{order_id}/fulfillments.json", headers=HEADERS
)
import asyncio
order_resp, fulfillment_resp = await asyncio.gather(order_task, fulfillment_task)
order = order_resp.json()["order"]
fulfillments = fulfillment_resp.json()["fulfillments"]
# 聚合结果
result = {
"order": {
"id": order["id"],
"status": order["financial_status"],
"total": order["total_price"],
"customer": order.get("customer", {}),
"line_items": order["line_items"],
},
"fulfillments": fulfillments
}
return [types.TextContent(type="text", text=json.dumps(result, ensure_ascii=False))]

模式三：语义化封装（LLM 最友好）

面向业务操作而非 API 端点设计工具，是对 LLM 最友好的方式：

# 语义化工具：业务操作层面的抽象
tools = [
types.Tool(
name="check_inventory_alert",
# 描述使用自然语言，说明什么时候该用这个工具
description="""
检查库存告警：查找库存量低于安全库存阈值的商品。
适合在以下情况调用：用户问"有哪些商品快卖完了"或"需要补货的有哪些"。
返回：商品名称、当前库存、安全库存阈值、建议补货量。
""",
inputSchema={
"type": "object",
"properties": {
"threshold_multiplier": {
"type": "number",
"default": 1.5,
"description": "低于 安全库存 × 这个倍数 则告警，默认1.5"
}
}
}
),
types.Tool(
name="process_refund",
description="""
处理退款：根据订单ID发起退款。
⚠️ 此操作不可逆，执行前必须向用户确认订单信息和退款金额。
""",
inputSchema={
"type": "object",
"properties": {
"order_id": {"type": "string"},
"reason": {
"type": "string",
"enum": ["customer_request", "quality_issue", "wrong_item", "other"]
},
"amount": {"type": "number", "description": "退款金额（为空则全额退款）"}
},
"required": ["order_id", "reason"]
}
)
]

工具描述的设计原则

好的工具描述能让 LLM 更准确地选择和调用工具：

API 认证的安全处理

import os
from functools import lru_cache
@lru_cache(maxsize=1)
def get_api_credentials() -> dict:
"""从环境变量读取 API 密钥，缓存避免重复读取"""
token = os.environ.get("SHOPIFY_ACCESS_TOKEN")
if not token:
raise ValueError("SHOPIFY_ACCESS_TOKEN 环境变量未设置")
return {
"X-Shopify-Access-Token": token,
"Content-Type": "application/json"
}

配置时通过 env 字段注入（不在代码中硬编码）：

{
"mcpServers": {
"shopify": {
"command": "python3",
"args": ["/path/to/shopify_server.py"],
"env": {
"SHOPIFY_ACCESS_TOKEN": "${SHOPIFY_TOKEN}",
"SHOPIFY_STORE_DOMAIN": "mystore.myshopify.com"
}
}
}
}

本节执行清单

• [ ] 根据使用频率和复杂度选择合适的封装模式
• [ ] 工具描述中说明适用场景、副作用和返回格式
• [ ] API 密钥通过环境变量注入，不硬编码
• [ ] 有副作用的工具（写/删除）在描述中添加警告标注

下一节：结果验证、幂等性与错误处理