MCP (Model Context Protocol) 详解 - 从概念到实践
什么是MCP？ MCP（Model Context Protocol）是一个让AI模型能够调用外部工具的协议。简单来说，它就像给AI装上了"手"，让AI能够使用各种外部服务。
举个例子 想象一下，你问AI：“今天北京天气怎么样？”
没有MCP：AI只能说"我不知道，我没有实时天气数据" 有MCP：AI可以调用天气API，然后告诉你"今天北京晴天，温度25度" MCP的核心概念 1. 工具（Tools） 定义：AI可以调用的外部功能 作用：让AI能做更多事情
比如：
查询天气 发送邮件 读取文件 调用API 2. 服务器（Server） 定义：提供工具的服务 作用：管理工具，处理AI的请求
3. 客户端（Client） 定义：AI模型本身 作用：调用工具，获取结果
4. 协议（Protocol） 定义：AI和工具之间的通信规则 作用：确保双方能正确理解对方
通信流程 AI助手 → MCP协议 → 工具服务器 → 外部服务 AI助手 ← MCP协议 ← 工具服务器 ← 外部服务 实际应用：天气查询工具 让我们实现一个简单的天气查询MCP工具。
1. 工具定义 { "name": "get_weather", "description": "查询指定城市的天气信息", "inputSchema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称" } }, "required": ["city"] } } 2. MCP服务器实现 # weather_mcp_server.py import json import requests from typing import Dict, Any class WeatherMCPServer: def __init__(self): # 这里使用免费的天气API，实际使用时需要申请API密钥 self.api_key = "your_api_key_here" self.base_url = "http://api.weatherapi.com/v1" def get_weather(self, city: str) -> Dict[str, Any]: """获取天气信息""" try: # 调用天气API url = f"{self.base_url}/current.json" params = { "key": self.api_key, "q": city, "lang": "zh" } response = requests.get(url, params=params) response.raise_for_status() data = response.json() # 提取天气信息 location = data["location"] current = data["current"] return { "success": True, "data": { "city": location["name"], "temperature": f"{current['temp_c']}°C", "condition": current["condition"]["text"], "humidity": f"{current['humidity']}%", "wind_speed": f"{current['wind_kph']} km/h", "feels_like": f"{current['feelslike_c']}°C" } } except Exception as e: return { "success": False, "error": str(e) } def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """处理MCP请求""" method = request.get("method") params = request.get("params", {}) request_id = request.get("id") if method == "tools/list": # 返回可用工具列表 return { "jsonrpc": "2.0", "id": request_id, "result": { "tools": [ { "name": "get_weather", "description": "查询指定城市的天气信息", "inputSchema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称" } }, "required": ["city"] } } ] } } elif method == "tools/call": # 调用工具 tool_name = params.get("name") arguments = params.get("arguments", {}) if tool_name == "get_weather": city = arguments.get("city") if not city: return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32602, "message": "缺少城市参数" } } result = self.get_weather(city) return { "jsonrpc": "2.0", "id": request_id, "result": result } else: return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32601, "message": f"工具 '{tool_name}' 不存在" } } else: return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32601, "message": f"方法 '{method}' 不存在" } } # 启动服务器 if __name__ == "__main__": server = WeatherMCPServer() # 模拟AI助手的请求 print("=== MCP天气查询服务器启动 ===") print("等待AI助手连接...") # 示例：处理工具列表请求 list_request = { "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} } response = server.handle_request(list_request) print("工具列表响应:") print(json.dumps(response, indent=2, ensure_ascii=False)) # 示例：处理天气查询请求 weather_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "get_weather", "arguments": { "city": "北京" } } } response = server.handle_request(weather_request) print("\n天气查询响应:") print(json.dumps(response, indent=2, ensure_ascii=False)) 3. 简化版本（不依赖外部API） # simple_weather_mcp.py import json from typing import Dict, Any class SimpleWeatherMCPServer: def __init__(self): # 模拟天气数据 self.weather_data = { "北京": { "temperature": "25°C", "condition": "晴天", "humidity": "45%", "wind_speed": "12 km/h" }, "上海": { "temperature": "28°C", "condition": "多云", "humidity": "65%", "wind_speed": "8 km/h" }, "广州": { "temperature": "30°C", "condition": "小雨", "humidity": "80%", "wind_speed": "15 km/h" } } def get_weather(self, city: str) -> Dict[str, Any]: """获取天气信息""" if city in self.weather_data: return { "success": True, "data": { "city": city, **self.weather_data[city] } } else: return { "success": False, "error": f"未找到城市 '{city}' 的天气信息" } def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]: """处理MCP请求""" method = request.get("method") params = request.get("params", {}) request_id = request.get("id") if method == "tools/list": return { "jsonrpc": "2.0", "id": request_id, "result": { "tools": [ { "name": "get_weather", "description": "查询指定城市的天气信息", "inputSchema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称（支持：北京、上海、广州）" } }, "required": ["city"] } } ] } } elif method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) if tool_name == "get_weather": city = arguments.get("city") if not city: return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32602, "message": "缺少城市参数" } } result = self.get_weather(city) return { "jsonrpc": "2.0", "id": request_id, "result": result } return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32601, "message": f"方法 '{method}' 不存在" } } # 测试服务器 def test_weather_server(): server = SimpleWeatherMCPServer() print("=== 测试MCP天气查询服务器 ===\n") # 测试1：获取工具列表 print("1. 获取可用工具列表") list_request = { "jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {} } response = server.handle_request(list_request) print("响应:") print(json.dumps(response, indent=2, ensure_ascii=False)) print() # 测试2：查询北京天气 print("2. 查询北京天气") weather_request = { "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { "name": "get_weather", "arguments": { "city": "北京" } } } response = server.handle_request(weather_request) print("响应:") print(json.dumps(response, indent=2, ensure_ascii=False)) print() # 测试3：查询不存在的城市 print("3. 查询不存在的城市") invalid_request = { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "get_weather", "arguments": { "city": "火星" } } } response = server.handle_request(invalid_request) print("响应:") print(json.dumps(response, indent=2, ensure_ascii=False)) if __name__ == "__main__": test_weather_server() 4. 运行结果示例 === 测试MCP天气查询服务器 === 1. 获取可用工具列表 响应: { "jsonrpc": "2.0", "id": 1, "result": { "tools": [ { "name": "get_weather", "description": "查询指定城市的天气信息", "inputSchema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称（支持：北京、上海、广州）" } }, "required": ["city"] } } ] } } 2. 查询北京天气 响应: { "jsonrpc": "2.0", "id": 2, "result": { "success": true, "data": { "city": "北京", "temperature": "25°C", "condition": "晴天", "humidity": "45%", "wind_speed": "12 km/h" } } } 3. 查询不存在的城市 响应: { "jsonrpc": "2.0", "id": 3, "result": { "success": false, "error": "未找到城市 '火星' 的天气信息" } } MCP的优势 标准化：所有工具都使用相同的接口 可扩展：可以轻松添加新工具 安全：可以控制AI能访问哪些工具 灵活：支持各种编程语言和平台 总结 MCP协议让AI变得更强大，能够：
