MCP介绍
MCP (Model Context Protocol) 详解：连接 AI 与外部世界的桥梁
作者：meimeitou
引言
随着人工智能技术的快速发展，AI 助手在我们的日常工作中扮演着越来越重要的角色。然而，大多数 AI 助手都存在一个共同的局限性：它们只能处理训练数据中的信息，无法实时访问外部数据源或执行具体的操作。为了解决这个问题，Anthropic 推出了 MCP（Model Context Protocol）——一个革命性的开放标准，为 AI 助手与外部世界的交互开辟了新的可能性。
什么是 MCP？
MCP（Model Context Protocol）是一个开放的通信协议，专门为连接 AI 助手与各种数据源和工具而设计。它提供了一个标准化的接口，使得 AI 应用程序能够安全、高效地访问外部资源，执行实际操作，从而大大扩展了 AI 助手的能力边界。
核心特性
- 标准化：统一的协议规范，确保不同系统间的兼容性
- 安全性：基于权限的访问控制，保证数据和操作的安全
- 扩展性：模块化设计，支持自定义工具和资源
- 实时性：支持实时数据访问和操作执行
MCP 的架构设计
客户端-服务器模式
MCP 采用经典的客户端-服务器架构：
┌─────────────────┐ MCP Protocol ┌─────────────────┐
│ MCP Client │◄──────────────────►│ MCP Server │
│ (AI Assistant) │ │ (External Tool) │
└─────────────────┘ └─────────────────┘MCP 客户端：
- Claude Desktop
- IDE 插件
- 自定义 AI 应用程序
MCP 服务器：
- 文件系统访问
- 数据库连接
- API 调用服务
- 计算工具
三大核心组件
1. Resources（资源）
Resources 是服务器可以公开的数据或内容，例如：
# 定义资源
resources = [
{
"uri": "file:///path/to/document.txt",
"name": "项目文档",
"description": "包含项目详细信息的文档",
"mimeType": "text/plain"
}
]2. Tools（工具）
Tools 是客户端可以调用的函数或操作：
# 定义工具
tools = [
{
"name": "search_web",
"description": "搜索互联网内容",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "搜索关键词"}
},
"required": ["query"]
}
}
]3. Prompts（提示）
Prompts 是预定义的提示模板，可以包含参数和上下文信息：
# 定义提示模板
prompts = [
{
"name": "code_review",
"description": "代码审查提示模板",
"arguments": [
{
"name": "language",
"description": "编程语言",
"required": True
}
]
}
]技术实现详解
传输层选择
MCP 支持多种传输方式：
- HTTP/WebSocket：适用于网络通信
- 标准输入/输出：适用于本地进程通信
- JSON-RPC 2.0：统一的消息格式
消息流程
Client Server
│ │
│──── Initialize ────────►│
│◄─── Capabilities ──────│
│ │
│──── List Resources ────►│
│◄─── Resources List ────│
│ │
│──── Call Tool ─────────►│
│◄─── Tool Result ───────│实际应用场景
1. 开发者工具集成
import subprocess
import json
from typing import Dict, Any, List
class GitMCPServer:
"""Git 操作 MCP 服务器"""
def __init__(self):
self.name = "git-mcp-server"
self.tools = [
{
"name": "git_status",
"description": "获取 Git 仓库状态",
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string"}
},
"required": ["path"]
}
},
{
"name": "git_commit",
"description": "提交代码更改",
"inputSchema": {
"type": "object",
"properties": {
"message": {"type": "string"},
"files": {"type": "array", "items": {"type": "string"}}
},
"required": ["message"]
}
}
]
async def git_status(self, path: str) -> Dict[str, Any]:
"""获取 Git 状态"""
try:
result = subprocess.run(
["git", "status", "--porcelain"],
cwd=path,
capture_output=True,
text=True,
check=True
)
return {
"status": "success",
"output": result.stdout,
"modified_files": result.stdout.strip().split('\n') if result.stdout.strip() else []
}
except subprocess.CalledProcessError as e:
return {
"status": "error",
"message": f"Git 命令执行失败: {e.stderr}"
}
async def git_commit(self, message: str, files: List[str] = None) -> Dict[str, Any]:
"""提交代码更改"""
try:
# 添加文件
if files:
subprocess.run(["git", "add"] + files, check=True)
else:
subprocess.run(["git", "add", "."], check=True)
# 提交更改
result = subprocess.run(
["git", "commit", "-m", message],
capture_output=True,
text=True,
check=True
)
return {
"status": "success",
"message": "提交成功",
"output": result.stdout
}
except subprocess.CalledProcessError as e:
return {
"status": "error",
"message": f"提交失败: {e.stderr}"
}2. 数据分析和可视化
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from typing import Dict, Any, List
import io
import base64
class DataAnalysisMCPServer:
"""数据分析 MCP 服务器"""
def __init__(self):
self.name = "data-analysis-mcp-server"
self.tools = [
{
"name": "analyze_csv",
"description": "分析 CSV 文件并生成统计报告",
"inputSchema": {
"type": "object",
"properties": {
"file_path": {"type": "string"},
"columns": {"type": "array", "items": {"type": "string"}}
},
"required": ["file_path"]
}
},
{
"name": "create_visualization",
"description": "创建数据可视化图表",
"inputSchema": {
"type": "object",
"properties": {
"file_path": {"type": "string"},
"chart_type": {"type": "string", "enum": ["bar", "line", "scatter", "histogram"]},
"x_column": {"type": "string"},
"y_column": {"type": "string"}
},
"required": ["file_path", "chart_type", "x_column", "y_column"]
}
}
]
async def analyze_csv(self, file_path: str, columns: List[str] = None) -> Dict[str, Any]:
"""分析 CSV 文件"""
try:
# 读取 CSV 文件
df = pd.read_csv(file_path)
# 如果指定了列，只分析这些列
if columns:
df = df[columns]
# 生成统计报告
analysis = {
"shape": df.shape,
"columns": df.columns.tolist(),
"data_types": df.dtypes.to_dict(),
"missing_values": df.isnull().sum().to_dict(),
"summary_statistics": df.describe().to_dict()
}
return {
"status": "success",
"analysis": analysis
}
except Exception as e:
return {
"status": "error",
"message": f"分析失败: {str(e)}"
}
async def create_visualization(self, file_path: str, chart_type: str,
x_column: str, y_column: str) -> Dict[str, Any]:
"""创建数据可视化"""
try:
# 读取数据
df = pd.read_csv(file_path)
# 创建图表
plt.figure(figsize=(10, 6))
if chart_type == "bar":
plt.bar(df[x_column], df[y_column])
elif chart_type == "line":
plt.plot(df[x_column], df[y_column])
elif chart_type == "scatter":
plt.scatter(df[x_column], df[y_column])
elif chart_type == "histogram":
plt.hist(df[x_column], bins=20)
plt.xlabel(x_column)
plt.ylabel(y_column)
plt.title(f"{chart_type.capitalize()} Chart: {x_column} vs {y_column}")
# 保存图表为 base64 编码
buffer = io.BytesIO()
plt.savefig(buffer, format='png')
buffer.seek(0)
image_base64 = base64.b64encode(buffer.read()).decode()
plt.close()
return {
"status": "success",
"chart_image": image_base64,
"chart_type": chart_type
}
except Exception as e:
return {
"status": "error",
"message": f"可视化创建失败: {str(e)}"
}3. 文件系统操作
import os
import shutil
from pathlib import Path
from typing import Dict, Any, List
class FileSystemMCPServer:
"""文件系统操作 MCP 服务器"""
def __init__(self, allowed_paths: List[str] = None):
self.name = "filesystem-mcp-server"
self.allowed_paths = allowed_paths or ["/tmp", "/home"]
self.tools = [
{
"name": "read_file",
"description": "读取文件内容",
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string"}
},
"required": ["path"]
}
},
{
"name": "write_file",
"description": "写入文件内容",
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"}
},
"required": ["path", "content"]
}
},
{
"name": "list_directory",
"description": "列出目录内容",
"inputSchema": {
"type": "object",
"properties": {
"path": {"type": "string"}
},
"required": ["path"]
}
}
]
def _validate_path(self, path: str) -> bool:
"""验证路径是否在允许的范围内"""
abs_path = os.path.abspath(path)
return any(abs_path.startswith(allowed) for allowed in self.allowed_paths)
async def read_file(self, path: str) -> Dict[str, Any]:
"""读取文件内容"""
if not self._validate_path(path):
return {
"status": "error",
"message": "路径不在允许的范围内"
}
try:
with open(path, 'r', encoding='utf-8') as file:
content = file.read()
return {
"status": "success",
"content": content,
"file_path": path
}
except Exception as e:
return {
"status": "error",
"message": f"读取文件失败: {str(e)}"
}
async def write_file(self, path: str, content: str) -> Dict[str, Any]:
"""写入文件内容"""
if not self._validate_path(path):
return {
"status": "error",
"message": "路径不在允许的范围内"
}
try:
# 确保目录存在
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, 'w', encoding='utf-8') as file:
file.write(content)
return {
"status": "success",
"message": f"文件已写入: {path}",
"bytes_written": len(content.encode('utf-8'))
}
except Exception as e:
return {
"status": "error",
"message": f"写入文件失败: {str(e)}"
}
async def list_directory(self, path: str) -> Dict[str, Any]:
"""列出目录内容"""
if not self._validate_path(path):
return {
"status": "error",
"message": "路径不在允许的范围内"
}
try:
items = []
for item in os.listdir(path):
item_path = os.path.join(path, item)
items.append({
"name": item,
"type": "directory" if os.path.isdir(item_path) else "file",
"size": os.path.getsize(item_path) if os.path.isfile(item_path) else None
})
return {
"status": "success",
"items": items,
"total_count": len(items)
}
except Exception as e:
return {
"status": "error",
"message": f"列出目录失败: {str(e)}"
}总结
MCP 作为连接 AI 与外部世界的桥梁，为开发者提供了强大而灵活的工具。通过 Python 实现 MCP 服务器，我们可以轻松地为 AI 助手添加各种功能，从简单的文件操作到复杂的数据分析和可视化。
关键要点：
- 安全第一：始终验证输入，限制访问权限
- 模块化设计：将不同功能分离到独立的模块中
- 错误处理：提供清晰的错误信息和恢复机制
- 性能优化：考虑异步处理和资源管理
- 文档完善：为每个工具提供详细的文档和示例
随着 MCP 生态系统的不断发展，Python 开发者有机会创建更多创新的 AI 集成解决方案，推动人工智能技术在各个领域的应用。