MCP 集成

模型上下文协议（MCP）通过标准化接口将AI代理连接到真实业务数据。

什么是MCP？

模型上下文协议是Anthropic提出的开放标准，用于将AI模型连接到外部数据源和工具。它为代理提供了一种统一的方式：

• - 查询数据库和API
• 访问文件和资源
• 执行工具和函数
• 跨会话维护上下文

MCP的重要性

在MCP之前：

• - 每次集成 = 自定义代码
• 不同API = 不同模式
• 工具间上下文丢失
• 安全性 = 每次集成临时处理

有了MCP：

• - 一个协议，多种集成
• 所有来源的标准模式
• 持久化上下文
• 内置安全模型

MCP架构

┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 客户端 │────▶│ 服务器 │────▶│ 资源 │
│ (代理) │ │ (MCP) │ │ (数据) │
└─────────────┘ └─────────────┘ └─────────────┘
│
┌──────┴──────┐
│ 工具 │
│ 提示 │
│ 资源 │
└─────────────┘

组件

1. MCP服务器

• - 暴露资源和工具
• 处理身份验证
• 管理连接

2. MCP客户端

• - 连接到服务器
• 发现能力
• 执行操作

3. 资源

• - 文件、数据库、API
• 读/写操作
• 更新订阅

4. 工具

• - 可执行函数
• 输入/输出模式
• 副作用

5. 提示

• - 可复用的提示模板
• 参数化
• 可组合

集成类型

1. 数据库集成

python

PostgreSQL的MCP服务器

from mcp import Server

server = Server(postgres-integration)

@server.resource(postgres://users)
async def get_users():
# 从数据库查询用户
return await db.query(SELECT * FROM users)

@server.tool(query_users)
async def query_users(filters: dict):
# 执行参数化查询
return await db.querywithfilters(filters)

2. API集成

python

REST API的MCP服务器

@server.resource(api://customers)
async def get_customers():
response = await httpx.get(https://api.example.com/customers)
return response.json()

@server.tool(create_customer)
async def create_customer(data: dict):
response = await httpx.post(
https://api.example.com/customers,
json=data
)
return response.json()

3. 文件系统集成

python

文件访问的MCP服务器

@server.resource(file://documents/{path})
async def read_document(path: str):
with open(fdocuments/{path}) as f:
return f.read()

@server.tool(write_document)
async def write_document(path: str, content: str):
with open(fdocuments/{path}, w) as f:
f.write(content)
return {status: written}

业务数据集成

广告数据

python

Google Ads MCP

@server.resource(ads://campaigns)
async def get_campaigns():
获取所有广告活动及其指标
campaigns = await adsclient.getcampaigns()
return normalize_campaigns(campaigns)

@server.tool(optimize_budget)
async def optimizebudget(campaignid: str):
自动调整广告活动预算
# 分析表现
# 调整支出分配
# 返回优化结果

分析数据

python

分析MCP

@server.resource(analytics://metrics)
async def get_metrics():
获取跨平台的标准化指标
return {
googleanalytics: await ga.getmetrics(),
mixpanel: await mixpanel.get_events(),
customevents: await custom.getevents()
}

@server.tool(query_analytics)
async def query_analytics(query: str):
自然语言分析查询
# 解析查询
# 跨平台执行
# 返回统一结果

CRM数据

python

Salesforce MCP

@server.resource(crm://leads)
async def get_leads():
从CRM获取线索
return await salesforce.query(SELECT Id, Name, Email FROM Lead)

@server.tool(create_lead)
async def create_lead(data: dict):
在CRM中创建新线索
lead = await salesforce.create(Lead, data)
return lead

最佳实践

1. 标准化

python

标准化来自不同来源的数据

def normalize_campaign(data, source):
schema = {
id: data.get(id) or data.get(campaign_id),
name: data.get(name) or data.get(campaign_name),
spend: data.get(spend) or data.get(cost),
impressions: data.get(impressions) or data.get(views),
clicks: data.get(clicks) or data.get(clicks_count),
source: source
}
return schema

2. 错误处理

python
@server.tool(risky_operation)
async def risky_operation(data: dict):
try:
result = await external_api.call(data)
return {success: True, data: result}
except APIError as e:
return {
success: False,
error: str(e),
suggestion: 请使用有效参数重试
}

3. 缓存

python
from functools import lru_cache
from datetime import datetime, timedelta

cache = {}

@server.resource(api://expensive-data)
async def getexpensivedata():
cache_key = expensive-data
cached = cache.get(cache_key)

if cached and cached[expires] > datetime.now():
return cached[data]

# 获取新数据
data = await expensiveapicall()
cache[cache_key] = {
data: data,
expires: datetime.now() + timedelta(hours=1)
}
return data

4. 安全性

python

验证输入

from pydantic import BaseModel

class QueryInput(BaseModel):
table: str
filters: dict
limit: int = 100

@server.tool(safe_query)
async def safe_query(input: QueryInput):
# 输入由Pydantic验证
# 防止SQL注入
return await db.query(input.table, input.filters, input.limit)

Claude桌面集成

json
// claudedesktopconfig.json
{
mcpServers: {
business-data: {
command: python,
args: [mcp_server.py],
env: {
DATABASE_URL: postgresql://...,
API_KEY: ...
}
}
}
}

常见MCP服务器

官方服务器

服务器	描述
filesystem	文件系统访问
postgres

PostgreSQL数据库 | | sqlite | SQLite数据库 | | github | GitHub API | | google-drive | Google Drive | | slack | Slack API |

自定义服务器

为以下内容创建自定义服务器：

• - 内部API
• 专有数据库
• 自定义工具
• 业务特定操作

调试

服务器日志

python
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(mcp_server)

@server.tool(debug_operation)
async def debug_operation(data: dict):
logger.debug(f输入: {data})
result = await process(data)
logger.debug(f输出: {result})
return result

连接问题

bash

测试MCP服务器

python -m mcp.server --debug

测试客户端连接

python -m mcp.client --url ws://localhost:8080

示例

查询多个数据源

python
@server.tool(crossplatformquery)
async def crossplatformquery(query: str):
跨多个平台查询
results = {}

# 查询每个平台
results[analytics] = await analytics.query(query)
results[crm] = await crm.query(query)
results[ads] = await ads.query(query)

# 合并结果
return merge_results(results)

自动洞察

python
@server.tool