1.MCP（模型上下文协议） #
开始使用Model Context Protocol (MCP)，让AI与数据无缝连接
2.什么是MCP？ #
MCP（Model Context Protocol）是一种开放协议，用于标准化应用程序如何向大语言模型（LLMs）提供上下文。简单来说，MCP就像是AI应用的USB-C接口。
2.1 类比理解 #
正如USB-C为设备连接各种外设和配件提供了标准化方式，MCP则为AI模型连接不同数据源和工具提供了标准化方法。MCP使您能够在LLMs之上构建智能体和复杂工作流，并将您的模型与世界连接起来。
3.MCP的核心价值 #
MCP为您提供：
- 📦 不断增长的预构建集成列表 - 您的LLM可以直接接入各种数据源
- 🔧 一种标准化的方式 - 为AI应用程序构建自定义集成
- 🌐 一个开放的协议 - 每个人都可以自由实现和使用
- 🔄 改变的自由度 - 在不同应用间无缝切换，随身携带你的上下文
4.快速体验：构建你的第一个MCP应用 #
下面提供两个完整的代码示例，帮助您快速理解MCP的工作原理。
4.1 MCP服务器 #
4.1.1 工作流程 #
4.1.2 server.py #
# 导入json模块，用于处理JSON数据
import json
# 导入sys模块，用于访问标准输入输出
import sys
# 从typing模块导入类型注解Dict和Any
from typing import Dict, Any
# 定义一个简单的MCP服务器类
class SimpleMCPServer:
# 构造函数，初始化服务器
def __init__(self):
# 定义服务器支持的工具列表
self.tools = {
# 工具1：读取文件
"read_file": {
# 工具描述
"description": "读取指定文件的内容",
# 工具参数定义
"parameters": {
"type": "object",
"properties": {
# 参数file_path，类型为字符串
"file_path": {
"type": "string",
"description": "要读取的文件路径",
}
},
# 必需参数
"required": ["file_path"],
},
},
# 工具2：获取天气
"get_weather": {
# 工具描述
"description": "获取指定城市的天气信息",
# 工具参数定义
"parameters": {
"type": "object",
"properties": {
# 参数city，类型为字符串
"city": {"type": "string", "description": "城市名称"}
},
# 必需参数
"required": ["city"],
},
},
}
# 处理客户端请求的方法
def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
# 获取请求的方法名
method = request.get("method")
# 获取请求参数，默认为空字典
params = request.get("params", {})
# 如果请求方法为tools/list
if method == "tools/list":
# 返回服务器支持的工具列表
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"result": {"tools": list(self.tools.values())},
}
# 如果请求方法为tools/call
elif method == "tools/call":
# 获取要调用的工具名称
tool_name = params.get("name")
# 获取工具调用参数
tool_args = params.get("arguments", {})
# 如果调用read_file工具
if tool_name == "read_file":
return self._read_file(tool_args, request.get("id"))
# 如果调用get_weather工具
elif tool_name == "get_weather":
return self._get_weather(tool_args, request.get("id"))
# 如果工具不存在
else:
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"error": {
"code": -32601,
"message": f"Method {tool_name} not found",
},
}
# 如果请求方法未知
else:
# 返回方法未找到的错误
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"error": {"code": -32601, "message": f"Method {method} not found"},
}
# 读取文件的私有方法
def _read_file(self, args: Dict[str, Any], request_id: Any) -> Dict[str, Any]:
# 获取文件路径参数
file_path = args.get("file_path")
try:
# 打开文件并读取内容
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
# 返回读取成功的结果
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"content": [
{
"type": "text",
"text": f"文件 {file_path} 的内容：\n{content}",
}
]
},
}
# 捕获异常，返回错误信息
except Exception as e:
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {"code": -32000, "message": f"读取文件失败：{str(e)}"},
}
# 获取天气信息的私有方法
def _get_weather(self, args: Dict[str, Any], request_id: Any) -> Dict[str, Any]:
# 获取城市参数，默认为北京
city = args.get("city", "北京")
# 定义模拟天气数据
weather_data = {
"北京": {"temperature": "25°C", "condition": "晴"},
"上海": {"temperature": "28°C", "condition": "多云"},
"广州": {"temperature": "30°C", "condition": "雨"},
}
# 获取指定城市的天气数据，若无则返回默认值
weather = weather_data.get(city, {"temperature": "20°C", "condition": "未知"})
# 返回天气信息
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"content": [
{
"type": "text",
"text": f"{city}的天气：{weather['condition']}，温度{weather['temperature']}",
}
]
},
}
# 主程序入口
if __name__ == "__main__":
# 创建服务器实例
server = SimpleMCPServer()
# 从标准输入循环读取每一行请求
for line in sys.stdin:
try:
# 解析JSON请求
request = json.loads(line.strip())
# 处理请求，获取响应
response = server.handle_request(request)
# 输出响应到标准输出
print(json.dumps(response, ensure_ascii=False))
# 刷新标准输出缓冲区
sys.stdout.flush()
# 捕获JSON解析错误
except json.JSONDecodeError:
# 输出解析错误响应
print(
json.dumps(
{
"jsonrpc": "2.0",
"id": None,
"error": {"code": -32700, "message": "Parse error"},
},
ensure_ascii=False,
)
)
# 刷新标准输出缓冲区
sys.stdout.flush()
4.2 MCP客户端 #
4.2.1 工作流程 #
4.2.2 client.py #
# 导入json模块，用于处理JSON数据
import json
# 导入subprocess模块，用于创建和管理子进程
import subprocess
# 导入sys模块，用于访问与Python解释器相关的变量和函数
import sys
# 从typing模块导入类型注解
from typing import Dict, Any, List
# 定义一个简单的MCP客户端类
class SimpleMCPClient:
# 构造函数，初始化客户端
def __init__(self, server_command: List[str]):
# 保存服务器启动命令
self.server_command = server_command
# 服务器进程对象初始化为None
self.server_process = None
# 工具列表初始化为空
self.tools = []
# 连接到MCP服务器
def connect(self):
try:
# 启动服务器进程，重定向标准输入、输出和错误
self.server_process = subprocess.Popen(
self.server_command, # 启动服务器进程
stdin=subprocess.PIPE, # 标准输入
stdout=subprocess.PIPE, # 标准输出
stderr=subprocess.PIPE, # 标准错误
text=True, # 设置为文本模式
bufsize=1, # 设置缓冲区大小
)
# 获取服务器支持的工具列表
self._get_tools()
# 打印连接成功信息
print(" 成功连接到MCP服务器")
# 返回True表示连接成功
return True
except Exception as e:
# 打印连接失败信息
print(f" 连接服务器失败：{str(e)}")
# 返回False表示连接失败
return False
# 获取服务器支持的工具列表
def _get_tools(self):
# 构造请求体，请求工具列表
request = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}
# 发送请求并获取响应
response = self._send_request(request)
# 如果响应中有result字段
if "result" in response:
# 提取工具列表
self.tools = response["result"].get("tools", [])
# 打印工具数量
print(f" 服务器支持 {len(self.tools)} 个工具")
# 遍历并打印每个工具的描述
for tool in self.tools:
print(f" - {tool.get('description', '未知工具')}")
# 向服务器发送请求并获取响应
def _send_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
try:
# 将请求对象序列化为JSON字符串
request_str = json.dumps(request, ensure_ascii=False)
# 写入到服务器进程的标准输入
self.server_process.stdin.write(request_str + "\n")
# 刷新输入缓冲区，确保数据发送
self.server_process.stdin.flush()
# 读取服务器进程的标准输出
response_line = self.server_process.stdout.readline()
# 如果没有读取到响应，返回错误信息
if not response_line:
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"error": {"code": -32000, "message": "服务器无响应"},
}
# 解析响应JSON字符串并返回
return json.loads(response_line.strip())
except Exception as e:
# 捕获异常并返回错误信息
return {
"jsonrpc": "2.0",
"id": request.get("id"),
"error": {"code": -32000, "message": f"请求失败：{str(e)}"},
}
# 调用服务器上的工具
def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
# 导入time模块，用于生成唯一ID
import time
# 使用当前时间戳生成唯一请求ID
request_id = int(time.time() * 1000)
# 构造工具调用请求体
request = {
"jsonrpc": "2.0",
"id": request_id,
"method": "tools/call",
"params": {"name": tool_name, "arguments": arguments},
}
# 打印调用工具的信息
print(f"🔧 调用工具：{tool_name}，参数：{arguments}")
# 发送请求并获取结果
result = self._send_request(request)
# 打印收到的响应
print(f"📥 收到响应：{result}")
# 返回结果
return result
# 交互模式，让用户手动调用工具
def interactive_mode(self):
# 打印交互模式提示信息
print("\n🎮 进入交互模式，输入 'quit' 退出")
print("可用命令：")
print(" read_file <文件路径>")
print(" get_weather <城市名>")
print(" list_tools")
print(" quit")
while True:
try:
# 获取用户输入的命令
command = input("\n请输入命令: ").strip()
# 如果输入quit，退出循环
if command == "quit":
break
# 如果输入list_tools，列出所有工具
elif command == "list_tools":
for i, tool in enumerate(self.tools, 1):
print(f"{i}. {tool.get('description', '未知工具')}")
# 如果输入read_file命令，调用read_file工具
elif command.startswith("read_file "):
file_path = command[10:].strip()
result = self.call_tool("read_file", {"file_path": file_path})
self._print_result(result)
# 如果输入get_weather命令，调用get_weather工具
elif command.startswith("get_weather "):
city = command[12:].strip()
result = self.call_tool("get_weather", {"city": city})
self._print_result(result)
# 其他未知命令
else:
print(" 未知命令，请重试")
except KeyboardInterrupt:
# 捕获Ctrl+C中断，退出循环
break
except Exception as e:
# 捕获其他异常并打印错误信息
print(f" 执行命令时出错：{str(e)}")
# 打印工具调用的结果
def _print_result(self, result: Dict[str, Any]):
# 打印原始结果
print(f"🔍 解析结果：{result}")
# 如果有result字段
if "result" in result:
# 获取内容列表
content = result["result"].get("content", [])
# 打印内容项数量
print(f" 找到 {len(content)} 个内容项")
# 遍历内容项并打印
for i, item in enumerate(content):
if item.get("type") == "text":
print(f"📄 内容 {i+1}: {item['text']}")
else:
print(f"📄 内容 {i+1}: {item}")
# 如果有error字段，打印错误信息
elif "error" in result:
print(f" 错误：{result['error']['message']}")
# 其他未知响应格式
else:
print(f"⚠️ 未知响应格式：{result}")
# 断开与服务器的连接
def disconnect(self):
# 如果服务器进程存在
if self.server_process:
# 终止服务器进程
self.server_process.terminate()
# 等待服务器进程结束
self.server_process.wait()
# 打印断开连接信息
print(" 已断开与服务器的连接")
# 主程序入口
if __name__ == "__main__":
# 检查命令行参数数量
if len(sys.argv) < 2:
# 打印用法提示
print("使用方法：python client.py <服务器脚本路径>")
print("示例：python client.py server.py")
# 退出程序
sys.exit(1)
# 获取服务器脚本路径
server_script = sys.argv[1]
# 创建客户端实例
client = SimpleMCPClient([sys.executable, server_script])
# 连接到服务器
if client.connect():
# 进入交互模式
client.interactive_mode()
# 断开连接
client.disconnect()5.如何运行示例 #
5.1 步骤1：保存代码文件 #
将上面的代码分别保存为 server.py 和 client.py
5.2 步骤3：在终端运行客户端 #
python client.py server.py5.3 步骤3：测试功能 #
在客户端交互模式中，您可以：
- 输入
list_tools查看可用工具 - 输入
read_file test.txt读取文件 - 输入
get_weather 北京获取天气信息 - 输入
quit退出