发布时间：2026-07-08 | 阅读时间：约 22 分钟 | 分类：MCP
摘要：手把手教你将现有企业 REST API 封装为 MCP Server，让 AI 助手直接调用内部系统。包含 FastMCP 框架实战、认证鉴权、错误处理和性能优化，提供完整可运行的代码示例。
为什么要把 API 包装成 MCP？
企业通常已有成熟的内部系统：CRM、ERP、HR 系统、工单平台等。这些系统通过 REST API 对外提供服务，但 AI 助手无法直接调用它们。
传统集成方式的问题：
- Function Calling 耦合：每个模型平台的 Function Calling 格式不同，OpenAI、Claude、Gemini 互不兼容
- 重复开发：为每个 AI 应用单独写适配层
- 难以维护：API 变更时需要修改多处适配代码
MCP 协议提供统一的抽象层：一次封装，处处可用。任何支持 MCP 的 AI 客户端（Claude Desktop、Cursor、Cline 等）都能直接调用你的企业 API。
MCP Server 架构设计
核心组件
┌─────────────────┐ ┌──────────────┐ ┌─────────────────┐
│ AI 客户端 │◄───►│ MCP Server │◄───►│ 企业 REST API │
│ (Claude/Cursor) │ │ (适配层) │ │ (CRM/ERP/HR) │
└─────────────────┘ └──────────────┘ └─────────────────┘
封装原则
- 薄层封装：MCP Server 只做协议转换，业务逻辑保持在原有 API
- 统一错误：将各种 API 错误转换为 MCP 标准错误格式
- 安全透传：认证信息在传输层处理，不暴露在提示词中
- 字段映射：将技术字段名映射为 AI 友好的描述
实战：封装企业内部 CRM API
场景假设
你的公司有一个 CRM 系统，提供以下 REST API：
GET /api/v1/customers?phone={phone}- 查询客户POST /api/v1/opportunities- 创建商机GET /api/v1/opportunities/{id}/followups- 查询跟进记录
目标：让销售团队的 AI 助手可以直接查询客户信息、创建商机。
Step 1：环境准备
mkdir mcp-crm-server && cd mcp-crm-server
python -m venv venv && source venv/bin/activate
pip install fastmcp httpx python-dotenv
touch .env server.py crm_client.py test_client.py
.env：
CRM_BASE_URL=https://crm.company.internal
CRM_API_KEY=sk-crm-live-key-xxx
MCP_SERVER_NAME=crm-mcp-server
Step 2：CRM 客户端封装
crm_client.py：
import os
import httpx
from typing import List, Dict, Optional
from dotenv import load_dotenv
load_dotenv()
class CRMClient:
"""CRM API 客户端 - 隔离底层 HTTP 细节"""
def __init__(self):
self.base_url = os.getenv("CRM_BASE_URL", "http://localhost:8080")
self.api_key = os.getenv("CRM_API_KEY", "")
self.client = httpx.Client(
base_url=self.base_url,
headers={"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"},
timeout=30.0,
)
def search_customer(self, phone: Optional[str] = None, name: Optional[str] = None) -> Dict:
"""根据手机号或姓名查询客户"""
params = {}
if phone:
params["phone"] = phone
if name:
params["name"] = name
try:
resp = self.client.get("/api/v1/customers", params=params)
resp.raise_for_status()
data = resp.json()
return {"success": True, "customers": data.get("items", [])}
except httpx.HTTPStatusError as e:
return {"success": False, "error": f"查询失败: {e.response.status_code}", "detail": e.response.text}
except Exception as e:
return {"success": False, "error": f"系统错误: {str(e)}"}
def create_opportunity(self, customer_id: str, title: str, value: float, stage: str = "初步接触") -> Dict:
"""创建销售商机"""
payload = {
"customer_id": customer_id,
"title": title,
"estimated_value": value,
"stage": stage,
"source": "AI助手",
}
try:
resp = self.client.post("/api/v1/opportunities", json=payload)
resp.raise_for_status()
return {"success": True, "opportunity": resp.json()}
except httpx.HTTPStatusError as e:
return {"success": False, "error": f"创建失败: {e.response.status_code}", "detail": e.response.text}
except Exception as e:
return {"success": False, "error": f"系统错误: {str(e)}"}
def get_followups(self, opportunity_id: str) -> Dict:
"""获取商机跟进记录"""
try:
resp = self.client.get(f"/api/v1/opportunities/{opportunity_id}/followups")
resp.raise_for_status()
return {"success": True, "followups": resp.json().get("items", [])}
except httpx.HTTPStatusError as e:
return {"success": False, "error": f"查询失败: {e.response.status_code}"}
except Exception as e:
return {"success": False, "error": f"系统错误: {str(e)}"}
def close(self):
self.client.close()
Step 3：MCP Server 实现
server.py：
import os
import json
from typing import Optional
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from crm_client import CRMClient
load_dotenv()
# 初始化 FastMCP
mcp = FastMCP(os.getenv("MCP_SERVER_NAME", "crm-mcp-server"))
# 全局 CRM 客户端（生产环境建议用连接池）
crm = CRMClient()
@mcp.tool()
async def search_customer(phone: Optional[str] = None, name: Optional[str] = None) -> str:
"""
查询客户信息。根据手机号或客户姓名搜索 CRM 系统中的客户档案。
Args:
phone: 客户手机号，如 13800138000
name: 客户姓名，如 张三（模糊匹配）
Returns:
客户信息 JSON，包含 id、姓名、手机号、等级、最近交易时间
"""
if not phone and not name:
return json.dumps({"error": "请提供手机号或姓名至少一个查询条件"}, ensure_ascii=False)
result = crm.search_customer(phone=phone, name=name)
if result["success"]:
customers = result["customers"]
if not customers:
return json.dumps({"message": "未找到匹配客户"}, ensure_ascii=False)
# 格式化输出，让 AI 更容易理解
formatted = []
for c in customers:
formatted.append({
"客户ID": c.get("id"),
"姓名": c.get("name"),
"手机号": c.get("phone"),
"等级": c.get("level", "普通"),
"最近交易": c.get("last_order_date", "无记录"),
"消费总额": f"¥{c.get('total_spent', 0):,.2f}",
})
return json.dumps({"找到客户": len(formatted), "客户列表": formatted}, ensure_ascii=False)
else:
return json.dumps({"error": result.get("error", "未知错误")}, ensure_ascii=False)
@mcp.tool()
async def create_opportunity(customer_id: str, title: str, estimated_value: float, stage: str = "初步接触") -> str:
"""
创建销售商机。在 CRM 中为指定客户创建新的销售机会/商机。
Args:
customer_id: 客户ID（从 search_customer 结果中获取）
title: 商机标题，如"Q3企业版软件采购"
estimated_value: 预估成交金额，单位元
stage: 销售阶段，可选：初步接触/需求确认/方案报价/商务谈判/赢单/输单
Returns:
创建结果，包含商机ID和后续操作指引
"""
# 输入校验
valid_stages = ["初步接触", "需求确认", "方案报价", "商务谈判", "赢单", "输单"]
if stage not in valid_stages:
return json.dumps({"error": f"无效的销售阶段。可选: {', '.join(valid_stages)}"}, ensure_ascii=False)
if estimated_value <= 0:
return json.dumps({"error": "预估金额必须大于0"}, ensure_ascii=False)
result = crm.create_opportunity(customer_id, title, estimated_value, stage)
if result["success"]:
opp = result["opportunity"]
return json.dumps({
"状态": "创建成功",
"商机ID": opp.get("id"),
"标题": opp.get("title"),
"预估金额": f"¥{estimated_value:,.2f}",
"当前阶段": stage,
"创建时间": opp.get("created_at"),
"下一步": "可以使用 add_followup 添加跟进记录"
}, ensure_ascii=False)
else:
return json.dumps({"error": result.get("error", "创建失败")}, ensure_ascii=False)
@mcp.tool()
async def get_followup_history(opportunity_id: str) -> str:
"""
查询商机跟进历史。获取指定商机的所有跟进记录和沟通日志。
Args:
opportunity_id: 商机ID
Returns:
跟进记录列表，按时间倒序排列
"""
result = crm.get_followups(opportunity_id)
if result["success"]:
followups = result["followups"]
if not followups:
return json.dumps({"message": "该商机暂无跟进记录"}, ensure_ascii=False)
formatted = []
for f in followups:
formatted.append({
"跟进时间": f.get("created_at"),
"跟进人": f.get("owner_name"),
"方式": f.get("contact_method", "未知"),
"内容摘要": f.get("summary", "无摘要")[:100],
})
return json.dumps({"记录数": len(formatted), "跟进历史": formatted}, ensure_ascii=False)
else:
return json.dumps({"error": result.get("error", "查询失败")}, ensure_ascii=False)
@mcp.resource("crm://status")
async def get_crm_status() -> str:
"""CRM 系统健康检查"""
try:
# 简单探测
result = crm.search_customer(phone="00000000000")
if "error" in result and "系统错误" in result.get("error", ""):
return "CRM 系统连接异常"
return "CRM 系统连接正常 | API 版本: v1 | 认证状态: 已认证"
except Exception as e:
return f"CRM 系统异常: {str(e)}"
if __name__ == "__main__":
# 启动 stdio 模式（适合 Claude Desktop）
mcp.run(transport='stdio')
Step 4：配置 Claude Desktop
在 Claude Desktop 的配置文件 claude_desktop_config.json 中添加：
{
"mcpServers": {
"crm": {
"command": "python",
"args": ["/absolute/path/to/mcp-crm-server/server.py"],
"env": {
"CRM_BASE_URL": "https://crm.company.internal",
"CRM_API_KEY": "sk-crm-live-key-xxx"
}
}
}
}
Step 5：测试验证
test_client.py：
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio
async def test():
server_params = StdioServerParameters(
command="python",
args=["server.py"],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 列出可用工具
tools = await session.list_tools()
print("可用工具:", [t.name for t in tools.tools])
# 测试查询客户
result = await session.call_tool("search_customer", {"phone": "13800138000"})
print("查询结果:", result)
if __name__ == "__main__":
asyncio.run(test())
运行测试：
python test_client.py
安全加固方案
1. 输入验证层
from pydantic import BaseModel, Field, validator
class OpportunityInput(BaseModel):
customer_id: str = Field(..., min_length=5, max_length=50)
title: str = Field(..., min_length=2, max_length=200)
estimated_value: float = Field(..., gt=0, le=100_000_000)
stage: str = Field(default="初步接触")
@validator('title')
def no_sql_injection(cls, v):
forbidden = [';', '--', 'drop', 'delete', 'update']
if any(f in v.lower() for f in forbidden):
raise ValueError('标题包含非法字符')
return v
2. 请求限流
from functools import wraps
import time
class RateLimiter:
def __init__(self, max_calls: int = 60, window: int = 60):
self.max_calls = max_calls
self.window = window
self.calls = []
def is_allowed(self) -> bool:
now = time.time()
self.calls = [c for c in self.calls if now - c < self.window]
if len(self.calls) < self.max_calls:
self.calls.append(now)
return True
return False
limiter = RateLimiter()
def rate_limited(func):
@wraps(func)
async def wrapper(*args, **kwargs):
if not limiter.is_allowed():
return json.dumps({"error": "请求过于频繁，请稍后再试"}, ensure_ascii=False)
return await func(*args, **kwargs)
return wrapper
@rate_limited
@mcp.tool()
async def search_customer(...):
...
3. 审计日志
import logging
from datetime import datetime
audit_logger = logging.getLogger("mcp.audit")
async def audit_log(tool_name: str, params: dict, result: str, user: str = "unknown"):
audit_logger.info(json.dumps({
"timestamp": datetime.now().isoformat(),
"user": user,
"tool": tool_name,
"params": params,
"result_preview": result[:200],
}, ensure_ascii=False))
性能优化技巧
| 优化项 | 方案 | 效果 |
|---|---|---|
| 连接池 | 使用 httpx.AsyncClient 复用连接 |
减少 TCP 握手开销 |
| 缓存 | 对客户信息做 5 分钟 Redis 缓存 | 降低 CRM 系统压力 |
| 并发 | FastMCP 天然支持异步 | 提升吞吐量 |
| 超时控制 | API 调用设置 10s 超时 | 防止慢请求拖垮系统 |
| 响应压缩 | 开启 gzip | 减少网络传输 |
常见问题 FAQ
Q1：现有 API 不是 REST 格式（如 SOAP、GraphQL）能封装吗？
A：完全可以。MCP Server 是协议无关的，你可以在 crm_client.py 中使用任何 HTTP 库或 SDK 调用原有系统，只要最终返回 JSON 给 MCP 层即可。
Q2：如何处理需要多步骤确认的敏感操作？
A：使用 UserProxyAgent 模式或在前端实现人工确认。也可以在 MCP Server 中设计两阶段提交：第一阶段返回预览信息，第二阶段通过 confirmed=true 参数执行。
Q3：MCP Server 如何部署为高可用服务？
A：使用 SSE 传输替代 stdio：
# 改为 HTTP 模式
mcp.run(transport='sse', host='0.0.0.0', port=8000)
然后用 Nginx 做负载均衡，后面挂多个 MCP Server 实例。
Q4：API 返回大量数据时如何优化？
A：1) 分页：在工具参数中添加 page 和 page_size；2) 摘要：只返回 AI 决策需要的字段；3) 流式：对于超大文本，使用 MCP 的 Resource 流式传输。
Q5：如何让 AI 理解复杂的业务规则？
A：在 system_message 或工具 description 中详细说明业务规则。对于特别复杂的规则，可以封装为独立的 check_business_rule 工具，让 AI 主动查询。
Q6：MCP Server 崩溃会影响 CRM 系统吗？
A：不会。MCP Server 是无状态的适配层，崩溃后重启即可。CRM 系统的稳定性由自身保障。
Q7：如何版本化管理 MCP Server？
A：在工具名中添加版本前缀（如 v2_search_customer），或在 mcp.resource 中暴露版本信息。建议遵循语义化版本（SemVer）。
Q8：多个团队各自封装 MCP Server，如何统一管理？
A：建立内部 MCP Registry（类似 npm registry），每个团队发布自己的 Server 包。统一配置中心管理各 Server 的连接信息和权限。
结语
将企业 API 封装为 MCP Server 是 AI 集成的主流趋势。它既保护了现有 IT 投资，又让 AI 能力快速渗透到业务一线。本文的 CRM 封装案例可以直接复用到你的实际系统中，建议从非关键查询接口开始试点，逐步扩展到写操作。
本文首发于 1630.top，转载请注明出处。