Python中实现MCP协议的开发指南
什么是 MCP 开发?为何开发者需要关注?
MCP(Model Context Protocol)是一项革命性技术,它为 AI 应用提供了标准化接口,让大语言模型(LLM)能够轻松访问外部工具、数据和系统。想象一下 USB 如何标准化了设备连接,MCP 就是 AI 世界的"万能接口"。
对开发者而言,MCP 解决了 AI 集成的最大痛点:如何让 AI 与你的系统安全、高效地交互。无需复杂的自定义接口,MCP 提供了一套统一标准,显著降低了开发成本和技术门槛。
MCP 开发能解决哪些实际问题?
1. 数据孤岛问题
传统 AI 只能处理对话框内的信息,而 MCP 让 AI 能够:
2. 工具调用壁垒
通过 MCP,AI 可以:
3. 安全与控制挑战
MCP 提供:
实战案例:用 Python 开发 MCP 服务器
案例一:企业知识库助手
# knowledge_base.py
from mcp.server.fastmcp import FastMCP
import sqlite3
mcp = FastMCP("knowledge-base")
@mcp.tool()
async def search_documents(query: str, department: str = None) -> str:
"""搜索企业文档库
Args:
query: 搜索关键词
department: 可选的部门筛选
"""
conn = sqlite3.connect('documents.db')
cursor = conn.cursor()
if department:
sql = "SELECT title, summary FROM documents WHERE content LIKE ? AND department = ?"
cursor.execute(sql, (f"%{query}%", department))
else:
sql = "SELECT title, summary FROM documents WHERE content LIKE ?"
cursor.execute(sql, (f"%{query}%",))
results = cursor.fetchall()
conn.close()
if not results:
return f"未找到与'{query}'相关的文档" # Changed double quotes to single quotes for clarity within f-string
return "\n\n".join([f"📄 {title}\n{summary}" for title, summary in results[:5]])
if __name__ == "__main__":
mcp.run(transport='stdio')
有了这个 MCP 服务器,用户可以询问:
"帮我找一下市场部关于第四季度营销策略的文档" "我需要技术团队关于云迁移的最新报告"
AI 将直接从企业知识库中检索相关信息,而不是返回"我无法访问您的内部系统"这样的回答。
案例二:全栈开发助手
# dev_assistant.py
from mcp.server.fastmcp import FastMCP
import subprocess
import os
mcp = FastMCP("dev-tools")
@mcp.tool()
async def run_code(language: str, code: str) -> str:
"""执行代码并返回结果
Args:
language: 编程语言(python, javascript, bash)
code: 要执行的代码
"""
if language == "python":
# 创建临时文件
with open("temp.py", "w") as f:
f.write(code)
# 执行代码并捕获输出
result = subprocess.run(
["python", "temp.py"],
capture_output=True,
text=True
)
# 清理临时文件
os.remove("temp.py")
# 返回执行结果
if result.returncode == 0:
return f"执行成功:\n{result.stdout}"
else:
return f"执行出错:\n{result.stderr}"
elif language == "javascript":
# 类似实现...
return "JavaScript 执行暂未实现" # Added placeholder return
elif language == "bash":
# 类似实现...
result = subprocess.run(
code,
shell=True, # Use shell=True for bash commands carefully
capture_output=True,
text=True
)
if result.returncode == 0:
return f"执行成功:\n{result.stdout}"
else:
return f"执行出错:\n{result.stderr}"
else:
return f"不支持的语言:{language}"
@mcp.tool()
async def search_codebase(query: str, file_type: str = None) -> str:
"""搜索代码库
Args:
query: 搜索关键词
file_type: 可选的文件类型筛选(如 py, js, css)
"""
cmd = ["grep", "-r", query, "."]
if file_type:
cmd = ["grep", "-r", query, "--include", f"*.{file_type}", "."]
try:
result = subprocess.run(cmd, capture_output=True, text=True, check=False) # Use check=False to handle no results
if not result.stdout:
return f"未找到包含'{query}'的代码" # Changed double quotes to single quotes
# 限制返回结果数量,避免过多
lines = result.stdout.strip().split("\n")[:10]
return "\n".join(lines)
except FileNotFoundError:
return "错误:未找到 grep 命令。请确保它已安装并在 PATH 中。"
except Exception as e:
return f"搜索代码库时出错: {str(e)}"
if __name__ == "__main__":
mcp.run(transport='stdio')
借助这个 MCP 服务器,开发者可以让 AI 帮助:
案例三:智能数据分析助手
# data_analyst.py
from mcp.server.fastmcp import FastMCP
import pandas as pd
import matplotlib.pyplot as plt
import io
import base64
import os # Import os
mcp = FastMCP("data-tools")
@mcp.tool()
async def analyze_csv(filepath: str, analysis_type: str) -> str:
"""分析 CSV 数据文件
Args:
filepath: CSV 文件路径
analysis_type: 分析类型(summary, correlation, group_by)
"""
try:
# 读取 CSV 文件
df = pd.read_csv(filepath)
if analysis_type == "summary":
# 返回数据摘要
summary = df.describe().to_string()
return f"数据摘要:\n{summary}"
elif analysis_type == "correlation":
# 计算相关性 (only on numeric columns)
numeric_df = df.select_dtypes(include='number')
if numeric_df.empty:
return "文件中没有数值列,无法计算相关性。"
corr = numeric_df.corr().to_string()
return f"相关性分析:\n{corr}"
elif analysis_type == "group_by":
# 假设文件中有 category 列
if "category" in df.columns:
# Group by 'category' and calculate mean for numeric columns
grouped = df.groupby("category").agg({col: 'mean' for col in df.select_dtypes(include='number').columns}).to_string()
return f"分组统计 (均值):\n{grouped}"
else:
return "文件中没有 category 列,无法进行分组分析"
else:
return f"不支持的分析类型:{analysis_type}"
except FileNotFoundError:
return f"错误:文件未找到 '{filepath}'"
except Exception as e:
return f"分析 CSV 时出错: {str(e)}"
@mcp.tool()
async def generate_chart(filepath: str, chart_type: str, x_column: str, y_column: str) -> str:
"""生成数据可视化图表
Args:
filepath: CSV 文件路径
chart_type: 图表类型(line, bar, scatter)
x_column: X 轴列名
y_column: Y 轴列名
"""
try:
# 读取 CSV 文件
df = pd.read_csv(filepath)
# 检查列是否存在
if x_column not in df.columns or y_column not in df.columns:
return f"错误:列 '{x_column}' 或 '{y_column}' 不在文件中。"
# 创建图表
plt.figure(figsize=(10, 6))
if chart_type == "line":
plt.plot(df[x_column], df[y_column])
elif chart_type == "bar":
# Bar chart often needs aggregation or specific handling if x_column is not unique categories
# For simplicity, assuming x_column represents categories or distinct values
plt.bar(df[x_column], df[y_column])
elif chart_type == "scatter":
plt.scatter(df[x_column], df[y_column])
else:
plt.close() # Close the figure if chart type is invalid
return f"不支持的图表类型:{chart_type}"
plt.xlabel(x_column)
plt.ylabel(y_column)
plt.title(f"{y_column} vs {x_column} ({chart_type.capitalize()} Chart)")
plt.tight_layout() # Adjust layout
# 将图表保存为图片文件
img_path = "chart.png"
plt.savefig(img_path)
plt.close() # Close the plot figure
return f"图表已生成并保存为:{img_path}"
except FileNotFoundError:
return f"错误:文件未找到 '{filepath}'"
except Exception as e:
plt.close() # Ensure figure is closed on error
return f"生成图表时出错: {str(e)}"
if __name__ == "__main__":
mcp.run(transport='stdio')
数据分析师可以请求:
"帮我分析这个销售数据的相关性" "根据产品类别生成一个条形图,展示各类别的销售额" "计算并展示客户年龄与购买金额的关系"
MCP 开发的核心优势
- 降低集成成本 传统 AI 集成需要定制开发,而 MCP 提供即插即用的体验,大幅降低工程成本。一个 MCP 服务器可以同时为多个 AI 客户端提供服务,实现一次开发多处使用。
- 增强 AI 能力 通过 MCP,AI 能够突破对话框限制,实现与真实系统的深度交互。不再仅限于回答问题,AI 可以执行任务、分析数据、控制系统。
- 安全与控制 MCP 设计中包含人工审核机制,确保 AI 的每一个关键操作都经过用户确认。数据可以在本地处理,无需上传到云端,提供更高的数据安全保障。
- 跨平台兼容 MCP 是厂商中立的开放协议,你的 MCP 服务器可以无缝连接不同的 AI 提供商,避免被单一平台锁定。
入门 MCP 开发的简单步骤
安装必要工具:
# 使用 uv 安装(推荐)
uv pip install mcp httpx
# 或使用传统 pip
pip install mcp httpx
创建你的第一个 MCP 服务器:
# hello_mcp.py
from mcp.server.fastmcp import FastMCP
# 初始化 MCP 服务器
mcp = FastMCP("hello-world")
@mcp.tool()
async def greet(name: str) -> str:
"""向用户问好
Args:
name: 用户名称
"""
return f"你好,{name}!欢迎使用 MCP!"
@mcp.tool()
async def calculate(expression: str) -> str:
"""计算数学表达式 (注意: eval() 有安全风险,仅用于演示)
Args:
expression: 数学表达式
"""
try:
# 警告: eval() 可以执行任意代码,非常不安全。
# 在生产环境中,应使用更安全的表达式求值库 (如 numexpr, ast.literal_eval)
allowed_chars = "0123456789+-*/(). "
if not all(c in allowed_chars for c in expression):
return "计算出错: 表达式包含不允许的字符"
result = eval(expression)
return f"计算结果: {result}"
except Exception as e:
return f"计算出错: {str(e)}"
if __name__ == "__main__":
# 启动服务器
mcp.run(transport='stdio')
运行并测试:
python hello_mcp.py
与 Claude Desktop 集成:
编辑 Claude Desktop 配置文件(macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
或 Windows: %APPDATA%\Claude\claude_desktop_config.json
),添加:
{
"mcpServers": {
"hello-world": {
"command": "python",
"args": ["/完整/路径/到/hello_mcp.py"]
}
}
}
适合 MCP 开发的场景
MCP 开发特别适合这些场景:
结语
MCP 开发正在改变 AI 应用的格局,它使 AI 从简单的对话机器人转变为能够实际行动和解决问题的助手。作为开发者,掌握 MCP 开发不仅能提升你的技术栈,还能为你的团队或客户创造真正有价值的 AI 解决方案。
无论你是希望增强现有系统的智能化水平,还是构建全新的 AI 驱动应用,MCP 都提供了一条清晰、高效的技术路径。开始 MCP 开发,探索 AI 与实际系统无缝融合的新可能!
README.md · randomuser1123/MCP开发指南 – Gitee.com
作者:m0_62919969