APPLICATION

理解模型上下文协议 (MCP)

在充分利用大型语言模型（LLMs）能力的过程中，开发者常常会遇到一个重大挑战：访问像Slack、GitHub或你自己的本地文件系统等流行工具提供的众多独特API。

admin

Mar 18, 2025 • 18 min read

在充分利用大型语言模型（LLMs）能力的过程中，开发者常常会遇到一个重大挑战：访问像Slack、GitHub或你自己的本地文件系统等流行工具提供的众多独特API。虽然LLMs可以通过各种工具连接到这些API，但这一过程通常需要编写代码来建立这些连接。

对于使用Cursor或Claude Desktop等桌面应用程序的用户来说，情况甚至更加受限，因为手动添加新工具是不可行的。

想象一下，如果可以无缝访问专门为你的现有桌面应用程序设计的一组预构建工具的世界会怎样？这就是模型上下文协议（MCP），一种革命性的方法，它允许您创建可以直接插入应用程序的可定制工具集。

MCP使LLMs能够指定各种命令，例如GitHub中的“获取存储库”或“评论PR”，提供无摩擦的体验。无论通过现有的桌面应用程序还是你自己的软件，MCP客户端都可以轻松与服务器通信以激活这些工具。

随着MCP服务器市场迅速兴起，无数公司渴望推出他们的解决方案。即将成为现实的是，这将赋予你从多样化的现成工具中进行选择的能力，增强你的LLM工作流程，并提高项目效率。🚀✨

1、什么是模型上下文协议（MCP）？

模型上下文协议（MCP）是一个具有状态、保留上下文的框架，旨在推动人类与AI代理之间的智能、多步骤交互。与传统API调用不同，后者将每次请求视为孤立事件，MCP引入了一个持久的、不断发展的上下文层 ，使AI系统能够保留记忆、动态学习并随着时间的推移自主行动。

根据modelcontextprotocol.io，MCP基于三个支柱：

状态性：维护会话特定和长期记忆。
互操作性：在模型、工具和数据源之间无缝工作。
以代理为中心的设计：优先考虑在定义边界内自主决策。

示例：

一个由MCP驱动的旅行代理 🧳 记住了你的预算、过敏情况以及过去的旅行反馈，跨对话计划个性化行程——无需重复自己！

2、模型上下文协议（MCP）的核心架构

理解客户端、服务器和LLMs之间的连接

模型上下文协议（MCP）具有灵活且可扩展的架构，可促进LLM应用程序与集成之间的无缝通信。本节概述了核心架构组件和概念。

2.1 概述

MCP运行在一个客户端-服务器架构上，其中：

主机：发起连接的LLM应用程序（如Claude Desktop或集成开发环境）。
客户端：与其对应的服务器保持1:1连接，运行在主机应用程序中。
服务器：向客户端提供必要的上下文、工具和提示。

2.2 服务器进程

服务器进程涉及以下组件：

主机：托管交互的应用程序。
传输层：负责在客户端和服务器之间进行通信的层。
MCP客户端：与MCP服务器通信的实体。
MCP服务器：管理工具、上下文并处理来自MCP客户端的请求的组件。

这种架构确保LLMs能够有效地访问和利用工具和资源，在各种应用中增强其能力。

3、模型上下文协议（MCP）的核心组件

3.1 协议层

协议层负责消息帧、链接请求与响应以及定义高级通信模式。

包括的关键类有：

协议
客户端
服务器

3.2 传输层

传输层管理客户端和服务器之间的实际通信，支持多种传输机制：

Stdio传输：利用标准输入/输出进行通信，非常适合本地进程。
HTTP带SSE传输：使用服务器发送事件进行服务器到客户端的消息传递，HTTP POST用于客户端到服务器的通信。

所有传输机制都依赖JSON-RPC 2.0进行消息交换。有关Model Context Protocol消息格式的详细信息，请参阅规范。

3.3 消息类型

MCP定义了几种关键消息类型：

请求：期望接收方响应。
结果：指示请求成功响应。
错误：表示请求失败。

4、MCP如何工作：技术深度解析

4.1 窗口管理

MCP使用动态上下文窗口，随着每次交互增长，存储：

用户偏好（例如，语言、语气）。
对话历史（之前的查询/响应）。
环境数据（例如，设备类型、位置 🌍）。

4.2 上下文嵌入与压缩

为了避免过载，MCP将非关键数据压缩为嵌入（例如，将10条消息的聊天总结为意图向量 🔢），同时保留关键细节。

4.3 状态化工作流

MCP支持多步工作流，其中代理：

记住过去的动作（例如，“用户已经上传了他们的ID”）。
适应策略（例如，如果用户离线，则从电子邮件切换到短信 📴）。
自我纠正使用反馈（例如，“用户不喜欢选项A；优先考虑选项B”）。

# 假设的MCP状态化工作流（来自官方文档）
class TravelAgent(MCPAgent):    
    def __init__(self, user_id):    
        self.context = load_context(user_id)  # 加载过去的交互    
        self.preferences = self.context.get("preferences", {})    
  
    def book_flight(self, query):    
        if self.preferences.get("class") == "economy":    
            return search_flights(query, budget=True)    
        else:    
            return search_flights(query)

5、为什么不直接给LLM访问API？

关于模型上下文协议（MCP）的一个常见问题是：“为什么我们需要自定义协议？LLMs不能简单地自学如何使用API吗？”

理论上，答案可能是肯定的。大多数公共API都附带有文档，概述了它们的功能。可以将这些文档提供给LLM，使其能够推导出实现目标所需的步骤。

然而，在实践中，这种方法往往效率低下。作为专注于提升用户体验的开发者，我们有必要优先考虑速度和响应性。通过以易于消费的形式呈现工具给LLM，我们显著减少了延迟并简化了整个过程。这种方法确保了用户交互更顺畅，结果更快。🏎️💨

6、MCP与传统API调用：变革者

为什么转变？ API就像快照 📸——非常适合静态任务。MCP是一段视频 🎥，捕捉用户的完整意图叙述。

这不是只是工具调用吗？

我经常遇到的一个关于模型上下文协议（MCP）的问题是，“这与工具调用有什么不同？”

工具调用是指LLM调用函数以执行现实世界任务的机制。在这种设置中，LLM与工具执行器协同工作，调用指定的工具并返回结果。典型的过程如下：

描述要调用的工具
发送结果
LLM
工具执行器

然而，这种交互通常发生在同一环境中，无论是单一服务器还是特定的桌面应用程序。

相比之下，MCP框架使LLMs能够访问来自另一个进程的工具，该进程可以是本地的也可以是远程服务器。结构如下：

🌐 MCP服务器
🖥️ MCP客户端
📝 描述要调用的工具
🔧 调用工具
📤 发送结果
🔄 返回结果
🤖 LLM
📜 MCP协议
⚙️ 工具执行器

关键区别在于MCP服务器与客户端的完全解耦。这种分离提供了更大的灵活性和可扩展性，增强了LLMs与外部工具的交互方式。🌐✨

7、MCP对自主AI的重要性

自主框架要求AI自主行动，而不仅仅是响应。MCP的重要性在于：

7.1 实现真正的自主权

代理现在可以：

基于历史数据做出决策（例如，医疗代理 🏥 回忆患者的过敏清单）。
无需人工干预即可链式任务（例如，“研究 → 草稿 → 编辑 → 发布”博客文章）。

7.2 协作智能

MCP允许代理共享上下文：

其他代理（例如，客户服务机器人 🤖将任务升级到人类代理 👩💻）。
外部工具（例如，将实时股票数据 📈集成到财务顾问的回答中）。

7.3 伦理护栏

可审计性：完整的上下文历史有助于追踪有偏见或不准确的输出。
隐私：敏感数据（例如，医疗记录）被隔离 🔒。

没有MCP，代理将缺乏连续性——就像厨师 🧑🍳在中途忘记食谱步骤一样！

7.4 实现长期自主性

持久记忆：代理记住用户的偏好（例如，“Alex讨厌垃圾邮件 📧”）。
目标链：执行多步任务（例如，研究 → 谈判 → 预订 商务旅行 ✈️）

8、连接生命周期

MPI中的连接生命周期对于管理客户端和服务器之间交互的状态和转换至关重要，确保在整个过程中通信和功能的健壮性。

这种对MCP组件的结构化方法提供了一个清晰的框架，以实现高效沟通和集成，使LLM应用程序蓬勃发展。

8.1 初始化

在初始化阶段，服务器和客户端之间发生以下步骤：

客户端发送一个包含协议版本及其能力的初始化请求。
服务器响应其自己的协议版本和能力。
客户端发送一个已初始化通知以确认连接建立成功。
连接现在可以使用，正常的消息交换开始。

8.2 消息交换

在初始化阶段之后，MCP支持以下通信模式：

请求-响应：客户端或服务器都可以发送请求，另一方将作出响应。
通知：任何一方都可以发送无需响应的一次性消息。

8.3 终止

连接可以通过任一方终止，这可以通过几种方式发生：

干净关闭：通过close()方法实现。
传输断开：当通信通道丢失时发生。
错误条件：遇到错误也可能导致终止。

8.4 错误处理

MCP定义了一组标准错误代码，以有效管理可能出现的问题：

enum ErrorCode {  
  // 标准JSON-RPC错误代码  
  ParseError = -32700,  
  InvalidRequest = -32600,  
  MethodNotFound = -32601,  
  InvalidParams = -32602,  
  InternalError = -32603  
}

此外，SDK和应用程序可以从-32000开始定义自己的自定义错误代码。

错误传播：

错误通过以下方式传达：

错误响应：返回给存在问题的请求。
错误事件：触发在传输上以通知错误。
协议级错误处理器：在MCP级别管理错误。

这种结构化的生命周期确保了客户端和服务器之间的稳健和高效通信，并在出现错误时优雅地处理它们。 🌐💼

9、代码实现

该实现展示了处理不同类型操作并通过各种传输协议的复杂多服务器设置。以下是系统的可视化表示：

安装所需依赖项

pip install mcp httpx langchain langchain-core langchain-community langchain-groq langchain-ollama langchain_mcp_adapters

在.env文件中设置groq api密钥

import os  
from dotenv import load_dotenv  
load_dotenv()

9.1 创建服务器

数学服务器

# math_server.py  
from mcp.server.fastmcp import FastMCP  
  
mcp = FastMCP("Math")  
  
@mcp.tool()  
def add(a: int, b: int) -> int:  
    """Add two numbers"""  
    return a + b  
  
@mcp.tool()  
def multiply(a: int, b: int) -> int:  
    """Multiply two numbers"""  
    return a * b  
  
if __name__ == "__main__":  
    mcp.run(transport="stdio")

天气服务器(weather.py)

from typing import Any  
import httpx  
from mcp.server.fastmcp import FastMCP  
  
# 初始化FastMCP服务器  
mcp = FastMCP("weather")  
  
# 常量  
NWS_API_BASE = "https://api.weather.gov"  
USER_AGENT = "weather-app/1.0"  
  
  
async def make_nws_request(url: str) -> dict[str, Any] | None:  
    """Make a request to the NWS API with proper error handling."""  
    headers = {  
        "User-Agent": USER_AGENT,  
        "Accept": "application/geo+json"  
    }  
    async with httpx.AsyncClient() as client:  
        try:  
            response = await client.get(url, headers=headers, timeout=30.0)  
            response.raise_for_status()  
            return response.json()  
        except Exception:  
            return None  
  
  
def format_alert(feature: dict) -> str:  
    """Format an alert feature into a readable string."""  
    props = feature["properties"]  
    return f"""  
Event: {props.get('event', 'Unknown')}  
Area: {props.get('areaDesc', 'Unknown')}  
Severity: {props.get('severity', 'Unknown')}  
Description: {props.get('description', 'No description available')}  
Instructions: {props.get('instruction', 'No specific instructions provided')}  
"""  
  
  
@mcp.tool()  
async def get_alerts(state: str) -> str:  
    """Get weather alerts for a US state.  
  
    Args:  
        state: Two-letter US state code (e.g. CA, NY)  
    """  
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"  
    data = await make_nws_request(url)  
  
    if not data or "features" not in data:  
        return "Unable to fetch alerts or no alerts found."  
  
    if not data["features"]:  
        return "No active alerts for this state."  
  
    alerts = [format_alert(feature) for feature in data["features"]]  
    return "\n---\n".join(alerts)  
  
  
@mcp.tool()  
async def get_forecast(latitude: float, longitude: float) -> str:  
    """Get weather forecast for a location.  
  
    Args:  
        latitude: Latitude of the location  
        longitude: Longitude of the location  
    """  
    # First get the forecast grid endpoint  
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"  
    points_data = await make_nws_request(points_url)  
  
    if not points_data:  
        return "Unable to fetch forecast data for this location."  
  
    # Get the forecast URL from the points response  
    forecast_url = points_data["properties"]["forecast"]  
    forecast_data = await make_nws_request(forecast_url)  
  
    if not forecast_data:  
        return "Unable to fetch detailed forecast."  
  
    # Format the periods into a readable forecast  
    periods = forecast_data["properties"]["periods"]  
    forecasts = []  
    for period in periods[:5]:  # Only show next 5 periods  
        forecast = f"""  
{period['name']}:  
Temperature: {period['temperature']}°{period['temperatureUnit']}  
Wind: {period['windSpeed']} {period['windDirection']}  
Forecast: {period['detailedForecast']}  
```*AI**: {item['text']}")  
        #                 else:  
        #                     # For simple text content  
        #                     print(f"**AI**: {message.content}")  
                              
        #     elif 'tools' in chunk and 'messages' in chunk['tools']:  
        #         for message in chunk['tools']['messages']:  
        #             if hasattr(message, 'name') and hasattr(message, 'content'):  
        #                 # Display tool response  
        #                 print(f"**Tool ({message.name})**: {message.content}")  
        return agent_response['messages'][-1].content  
if __name__ == "__main__":  
    #user_question = "what is the weather in california?"  
    #user_question = "what's (3 + 5) x 12?"  
    #user_question = "what's the weather in seattle?"  
    user_question = "what's the weather in NYC?"  
    response = asyncio.run(run_app(user_question=user_question))  
    print(response)

在启动客户端之前，请确保天气服务器已启动并运行。

python weather.py

启动客户端

python langchain_mcp_multiserver.py

响应：“what’s (3 + 5) x 12?”

(3 + 5) 的结果是 8，8 x 12 是 96。

响应 : “what’s the weather in NYC?”

看起来你提供了来自国家气象局（NWS）的各种地区的天气警报纽约州、佛蒙特州和马萨诸塞州的部分地区。  
  
以下是每个警报的内容：  
  
**洪水警报**  
  
* NWS 已经在纽约州各地发布了多个洪水观察，包括：  
        + 北部圣劳伦斯；北部富兰克林；东部克林顿；东南部圣劳伦斯；南部富兰克林；西部克林顿；西部埃塞克斯；西南部圣劳伦斯；格兰德艾尔；西部富兰克林；奥尔巴尼；埃塞克斯；西部切坦登；拉莫伊尔；卡莱东尼亚；华盛顿；西部阿迪森；橙县；西部鲁特兰；东部富兰克林；东部切坦登；东部阿迪森；东部鲁特兰；西部温莎；东部温莎  
        + 北部赫基默；汉密尔顿；南部赫基默；南部富尔顿；蒙哥马利；北部萨拉托加；北部沃伦；北部华盛顿；北部富尔顿；东南沃伦；南部华盛顿；本宁顿；西部温德姆；东部温德姆  
* NWS 还为佛蒙特州部分地区发布了洪水观察，包括：  
        + 纽约州北部和佛蒙特州北部及中部  
  
**冰坝警报**  
  
* NWS 警告说一些地区可能会出现冰坝，包括：  
        + 本宁顿；西部温德姆；东部温德姆  
        + 佛蒙特州南部，本宁顿和温德姆县  
        + 纽约州中部，赫基默县  
        + 纽约州北部，汉密尔顿、蒙哥马利、富尔顿、赫基默、沃伦、华盛顿县  
  
**其他警报**  
  
* NWS 发布了多条警告，称大雨和融雪导致小河流洪水。  
* 还有孤立的冰坝警报，可能会进一步增加洪水风险。  
  
请务必关注您所在地区的天气状况，并遵循当地当局的指示。如果您计划进行户外活动，请准备好应对不断变化的天气条件，并采取必要的预防措施以确保安全。  
看起来你提供了来自国家气象局（NWS）的各种地区的天气警报纽约州、佛蒙特州和马萨诸塞州的部分地区。  
  
以下是每个警报的内容：  
  
**洪水警报**  
  
* NWS 已经在纽约州各地发布了多个洪水观察，包括：  
        + 北部圣劳伦斯；北部富兰克林；东部克林顿；东南部圣劳伦斯；南部富兰克林；西部克林顿；西部埃塞克斯；西南部圣劳伦斯；格兰德艾尔；西部富兰克林；奥尔巴尼；埃塞克斯；西部切坦登；拉莫伊尔；卡莱东尼亚；华盛顿；西部阿迪森；橙县；西部鲁特兰；东部富兰克林；东部切坦登；东部阿迪森；东部鲁特兰；西部温莎；东部温莎  
        + 北部赫基默；汉密尔顿；南部赫基默；南部富尔顿；蒙哥马利；北部萨拉托加；北部沃伦；北部华盛顿；北部富尔顿；东南沃伦；南部华盛顿；本宁顿；西部温德姆；东部温德姆  
* NWS 还为佛蒙特州部分地区发布了洪水观察，包括：  
        + 纽约州北部和佛蒙特州北部及中部  
  
**冰坝警报**  
  
* NWS 警告说一些地区可能会出现冰坝，包括：  
        + 本宁顿；西部温德姆；东部温德姆  
        + 佛蒙特州南部，本宁顿和温德姆县  
        + 纽约州中部，赫基默县  
        + 纽约州北部，汉密尔顿、蒙哥马利、富尔顿、赫基默、沃伦、华盛顿县  
  
**其他警报**  
  
* NWS 发布了多条警告，称大雨和融雪导致小河流洪水。  
* 还有孤立的冰坝警报，可能会进一步增加洪水风险。  
  
请务必关注您所在地区的天气状况，并遵循当地当局的指示。如果您计划进行户外活动，请准备好应对不断变化的天气条件，并采取必要的预防措施以确保安全。

注意这里，MCP 客户端能够根据所提问题连接到相应的服务器。我们并未明确提及任何路由逻辑。

10、MCP 的未来

根据官方文档，即将推出的功能包括：

跨平台上下文同步：统一不同应用程序中的上下文（例如，您的电子邮件 ✉️、Slack 💬 和 CRM 📊）。
基于上下文的安全性：根据用户行为动态调整权限。
自我优化：代理自行完善其上下文管理规则 🔄。

关键要点：

对于复杂、不断发展的任务，MCP > APIs。
关键在于代理 🤖 主动行动，而不仅仅是被动反应。
平衡权力与安全 通过可审计、模块化的上下文。

该架构特别适用于：

多模态 AI 应用程序
复杂的工作流编排
分布式 AI 系统
实时数据处理应用程序

11、结束语

此处展示的 MCP 实现展示了构建复杂 AI 应用程序的强大、可扩展且灵活的架构。其模块化设计和支持多种传输协议使其成为复杂 AI 系统的绝佳选择。

原文链接：Understanding Model Context Protocol: A Deep Dive into Multi-Server LangChain Integration

汇智网翻译整理，转载请标明出处