MCP Business Integration

MCP Integration

模型 上下文 Protocol (MCP) connects AI 代理s to real business data through normalized interfaces.

What is MCP?

模型 上下文 Protocol is Anthropic's open standard for connecting AI 模型s to external data sources and 工具s. It provides a unified way for 代理s to:

查询 databases and APIs 访问 files and resources 执行 工具s and functions MAIntAIn 上下文 across 会话s Why MCP Matters

Before MCP:

Each integration = custom code Different APIs = different patterns 上下文 lost between 工具s Security = ad-hoc per integration

With MCP:

One protocol, many integrations Standard patterns for all sources Persistent 上下文 Built-in security 模型 MCP Architecture ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ 命令行工具ent │────▶│ Server │────▶│ Resource │ │ (代理) │ │ (MCP) │ │ (Data) │ └─────────────┘ └─────────────┘ └─────────────┘ │ ┌──────┴──────┐ │ 工具s │ │ Prompts │ │ Resources │ └─────────────┘

组件s

• MCP Server

Exposes resources and 工具s Handles authentication Manages connections

• MCP 命令行工具ent

Connects to servers Discovers capabilities 执行s operations

• Resources

Files, databases, APIs Read/write operations Subscriptions for 更新s

• 工具s

Executable functions 输入/输出 模式s Side effects

• Prompts

Reusable prompt templates Parameterized Composable Integration Types

• Database Integration

# MCP Server for PostgreSQL from mcp 导入 Server

server = Server("postgres-integration")

@server.resource("postgres://users") a同步 def 获取_users(): # 查询 users from database return awAIt db.查询("SELECT * FROM users")

@server.工具("查询_users") a同步 def 查询_users(过滤器s: dict): # 执行 parameterized 查询 return awAIt db.查询_with_过滤器s(过滤器s)

• API Integration

# MCP Server for REST API @server.resource("API://customers") a同步 def 获取_customers(): 响应 = awAIt httpx.获取("https://API.example.com/customers") return 响应.json()

@server.工具("创建_customer") a同步 def 创建_customer(data: dict): 响应 = awAIt httpx.post( "https://API.example.com/customers", json=data ) return 响应.json()

• File 系统 Integration

# MCP Server for file 访问 @server.resource("file://documents/{path}") a同步 def read_document(path: str): with open(f"documents/{path}") as f: return f.read()

@server.工具("write_document") a同步 def write_document(path: str, content: str): with open(f"documents/{path}", "w") as f: f.write(content) return {"状态": "written"}

Business Data Integration Ads Data # Google Ads MCP @server.resource("ads://campAIgns") a同步 def 获取_campAIgns(): """获取 all ad campAIgns with 指标""" campAIgns = awAIt ads_命令行工具ent.获取_campAIgns() return normalize_campAIgns(campAIgns)

@server.工具("优化_bud获取") a同步 def 优化_bud获取(campAIgn_id: str): """Automatically adjust campAIgn bud获取""" # Analyze performance # Adjust spend allocation # Return optimization 结果s

分析 Data # 分析 MCP @server.resource("分析://指标") a同步 def 获取_指标(): """获取 normalized 指标 across 平台s""" return { "google_分析": awAIt ga.获取_指标(), "mixpanel": awAIt mixpanel.获取_事件(), "custom_事件": awAIt custom.获取_事件() }

@server.工具("查询_分析") a同步 def 查询_分析(查询: str): """Natural language 分析 查询""" # 解析 查询 # 执行 across 平台s # Return unified 结果s

CRM Data # Salesforce MCP @server.resource("crm://leads") a同步 def 获取_leads(): """获取 leads from CRM""" return awAIt salesforce.查询("SELECT Id, Name, EmAIl FROM Lead")

@server.工具("创建_lead") a同步 def 创建_lead(data: dict): """创建 new lead in CRM""" lead = awAIt salesforce.创建("Lead", data) return lead

Best Practices

• Normalization

# Normalize data from different sources def normalize_campAIgn(data, source): 模式 = { "id": data.获取("id") or data.获取("campAIgn_id"), "name": data.获取("name") or data.获取("campAIgn_name"), "spend": data.获取("spend") or data.获取("cost"), "impressions": data.获取("impressions") or data.获取("views"), "命令行工具cks": data.获取("命令行工具cks") or data.获取("命令行工具cks_count"), "source": source } return 模式

• Error Handling

@server.工具("risky_operation") a同步 def risky_operation(data: dict): try: 结果 = awAIt external_API.call(data) return {"成功": True, "data": 结果} except APIError as e: return { "成功": False, "error": str(e), "suggestion": "Try agAIn with valid parameters" }

• Caching

from func工具s 导入 lru_缓存 from datetime 导入 datetime, timedelta