什么是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的优势

  1. 标准化：所有工具都使用相同的接口
  2. 可扩展：可以轻松添加新工具
  3. 安全：可以控制AI能访问哪些工具
  4. 灵活：支持各种编程语言和平台

总结

MCP协议让AI变得更强大，能够：

  • 查询实时信息（如天气、股票）
  • 执行操作（如发送邮件、保存文件）
  • 访问外部服务（如数据库、API）

通过这个简单的天气查询例子，我们可以看到MCP是如何工作的。AI助手通过MCP协议调用天气工具，获取实时天气信息，然后回答用户的问题。

本文作者：一缘
转载请注明出处