什么是MCP协议

2025年4月1日 · 阅读需 8 分钟

Front And Rear End Engineers @ gitee

在现代 AI 的发展中，模型与外部数据源和工具的交互变得越来越复杂。为了解决这个问题，Model Context Protocol (MCP) 被提出，它是一种开放标准，旨在为 AI 模型与外部资源之间提供统一的连接方式。由 Anthropic 于 2024 年推出，MCP 希望能够像 USB-C 接口一样成为 AI 系统与各种数据源和工具之间的“通用接口。这意味着，不同的 AI 应用可以轻松与多种数据源、工具进行即插即用的连接，无需再为每种不同的数据源开发独立的集成方法。

MCP 协议通过为 AI 模型与外部数据源和工具提供统一的连接接口，简化了应用与外部资源的集成过程。其标准化的协议架构、双向通信能力以及良好的扩展性，使其成为 AI 应用开发中的一个重要工具。随着 MCP 协议的进一步发展和普及，它有望成为 AI 领域的USB-C 接口，为各种 AI 应用提供更加智能、灵活的连接方式。

MCP 的技术架构

MCP 的技术架构可以类比为计算机上的USB-C 集线器。USB-C 接口能够让不同的设备通过统一的接口进行连接，MCP 则为 AI 模型与外部数据源和工具提供了类似的通道。

MCP 协议由三种主要角色组成：

MCP Host（主机）：运行大型语言模型（LLM）的应用程序，例如 Claude 桌面版、IDE 插件或聊天机器人。Host 是与外部数据源或工具交互的核心，通常需要从外部获取上下文数据或执行操作。
MCP Server（服务器）：每个 MCP Server 提供特定的功能或数据访问能力。例如，它可能是一个处理文件系统读写、数据库查询或发送 Slack 消息的服务。Server 会封装与实际数据源的交互逻辑，并对外暴露统一的 MCP 接口。
MCP Client（客户端）：Client 充当 Host 和 Server 之间的中介，负责将 Host 的请求转发给 Server，并将 Server 的响应返回给 Host。每个 Client 与一个特定的 Server 建立 1:1 的连接。

通过这种架构，MCP 协议消除了传统上每种数据源都需要单独集成的繁琐步骤，使得 AI 应用能够通过统一的接口与各种外部资源进行交互。

核心原理与协议层

MCP 协议基于标准的客户端-服务器架构，其中 Host、Client 和 Server 分别承担不同的角色。通信过程遵循 JSON-RPC 等通用 Web 协议来实现双向通信。具体来说，Host 通过 MCP Client 调用 Server 提供的服务或方法（例如读取文件、执行搜索等），而 Server 完成相应操作后将结果返回给 Host。

整个过程采用 JSON 格式的消息进行数据传输，这样做的好处是确保了上下文在多次交互中得以保持，实现了“连续对话”的功能，而不仅仅是一次性 API 调用。通过这种机制，AI 模型可以与外部工具进行深度的交互，提供更加智能化和人性化的服务。

双向交互与数据传输

与传统的 API 通信不同，MCP 协议支持双向交互。这意味着，除了模型可以请求数据外，Server 也可以根据需要向 Host 发出信息。例如，在某些情况下，Server 可能会要求用户授权操作，或向 Host 提供重要的提示信息。

数据传输机制方面，MCP 协议支持多种传输层实现，如 HTTP 和 WebSocket。这使得它能够适应不同的应用需求，并且具有良好的扩展性。未来，MCP 协议还计划支持无状态Serverless场景，以适应云环境中的使用需求。

MCP 封装的协议

MCP 协议本身并不直接与特定的通信协议绑定，而是通过标准化的 JSON 消息格式和通用的通信协议（如 JSON-RPC、HTTP、WebSocket 等）来实现与外部工具的交互。因此，MCP 协议能够封装多种不同类型的协议，使得 AI 应用能够与各种不同的数据源和工具进行连接。

这些协议的封装使得开发者可以更加专注于 AI 模型的开发，而无需过多关注不同工具之间的集成问题。例如，开发者只需要实现与 MCP 协议兼容的 Client，就能够与不同类型的 Server 进行无缝对接。

MCP 的优势与应用前景

统一接口：MCP 提供了一个统一的接口，解决了不同数据源和工具之间的兼容问题，使得 AI 应用能够更加灵活地与外部资源进行交互。 MCP 协议为模型与工具之间的交互提供了标准化的接口，开发者无需处理复杂的集成逻辑。这简化了开发过程，使得模型能够更快速、方便地与外部数据源和工具进行交互，降低了开发的复杂度和成本。
降低集成成本：开发者不再需要为每种工具或数据源编写单独的集成代码，通过 MCP 协议，AI 模型能够轻松接入多种服务，减少了开发和维护的工作量。
双向通信：MCP 协议的双向通信机制使得 AI 应用可以与外部工具和数据源进行更为复杂和智能的交互，为未来的智能化应用提供了更加广阔的可能性。
高度扩展性：MCP 支持多种数据传输机制和传输层协议，具有良好的扩展性，能够适应不同的应用需求。MCP 协议不仅提供简单的数据传输，还允许模型主动查询数据库、调用 API、执行代码等。这极大地扩展了 AI 模型的能力，帮助模型与外部环境进行更为复杂的交互。
跨平台：MCP 协议设计为跨平台、跨应用的标准接口，使得不同的 AI 系统和工具能够通过 MCP 轻松对接。开发者无需为每种工具单独编写集成代码，只需遵循 MCP 协议，就可以与多种数据源和工具无缝对接。

创建一个MCP Server

import requests
from bs4 import BeautifulSoup
from fastmcp import FastMCP
from fastapi import FastAPI, HTTPException
from typing import Dict, Optional

# 和风天气API配置
BASE_URL = "https://api.qweatherapi.com/"
# 从环境变量读取API密钥（安全起见，不硬编码）
API_KEY = os.getenv("QWEATHER_API_KEY")

# 初始化 FastMCP server
app = FastMCP("天气查询 MCP 服务", log_level="ERROR")

def get_city_location(city_name: str) -> Optional[str]:
    """
    根据城市名获取城市ID（和风天气需要城市ID查询天气）
    """
    if not API_KEY:
        print("错误：未配置QWEATHER_API_KEY环境变量")
        return None
    
    url = f"{BASE_URL}geo/v2/city/lookup"
    params = {
        "key": API_KEY,
        "location": city_name,
        "lang": "zh",
        "type": "city"
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()

        if data.get("code") != "200" or not data.get("location"):
            return None
        
        # 返回第一个匹配的城市ID
        return data["location"][0]["id"], data["location"][0]["lat"], data["location"][0]["lon"]
    
    except requests.exceptions.RequestException as e:
        print(f"获取城市位置失败：{str(e)}")
        return None
        
# 请求天气
@app.tool(description="请根据城市名称查询天气")
async def get_weather_request(city: str) -> str | None:
    """
    核心天气查询逻辑
    :param city: 城市路径，如 上海
    :return: 天气信息字符串
    """
    weather_data = {}
    
    city_id, latitude, longitude = get_city_location(city)
    if not city_id:
        print(f"未查询到「{city}」的天气数据，请检查城市名是否正确")
        return None
    
    # 1. 查询实时天气
    now_url = f"{BASE_URL}v7/weather/now"
    now_params = {"key": API_KEY, "location": city_id, "lang": "zh"}
    try:
        now_resp = requests.get(now_url, params=now_params, timeout=10)
        now_resp.raise_for_status()
        now_data = now_resp.json()
        
        if now_data.get("code") == "200":
            weather_data["now"] = {
                "temp": now_data["now"]["temp"],
                "feels_like": now_data["now"]["feelsLike"],
                "text": now_data["now"]["text"],
                "wind_scale": now_data["now"]["windScale"]
            }
            weather_data["city_name"] = city
        else:
            print(f"实时天气查询失败：{now_data.get('code')}")
            return None
    
    except requests.exceptions.RequestException as e:
        print(f"实时天气请求失败：{str(e)}")
        return None
    
    # 2. 查询未来3天预报
    forecast_url = f"{BASE_URL}v7/weather/3d"
    forecast_params = {"key": API_KEY, "location": city_id, "lang": "zh"}
    try:
        forecast_resp = requests.get(forecast_url, params=forecast_params, timeout=10)
        forecast_resp.raise_for_status()
        forecast_data = forecast_resp.json()
        
        if forecast_data.get("code") == "200":
            weather_data["forecast"] = forecast_data["daily"][:3]  # 取前3天
        else:
            print(f"预报查询失败：{forecast_data.get('code')}")
            return None
    
    except requests.exceptions.RequestException as e:
        print(f"预报请求失败：{str(e)}")
        return None
    
    # 3. 查询空气质量
    aqi_url = f"{BASE_URL}airquality/v1/current/{latitude}/{longitude}"
    aqi_params = {"key": API_KEY}
    try:
        aqi_resp = requests.get(aqi_url, params=aqi_params, timeout=10)
        aqi_resp.raise_for_status()
        aqi_data = aqi_resp.json()
        
        weather_data["aqi"] = {
            "value": aqi_data["indexes"][0]["aqi"],
            "level": aqi_data["indexes"][0]["category"]
        }
    except requests.exceptions.RequestException:
        # 空气质量非必需，失败不影响主功能
        weather_data["aqi"] = {"value": "暂无", "level": "暂无"}
    
    return weather_data


if __name__ == "__main__":
    # 启动日志
    print("天气查询 MCP 服务启动成功！")
    print(f"工具名称：get_weather")
    print(f"服务地址：http://0.0.0.0:9000/tool/get_weather?city=城市名")
    app.run(transport="sse", host="0.0.0.0", port=9000)

测试结果

MCP 的技术架构​

核心原理与协议层​

双向交互与数据传输​

MCP 封装的协议​

MCP 的优势与应用前景​

创建一个MCP Server​