技术详解丨深度分析MCP工作原理，附代码实现（下）

01 MCP 工作原理：技术深度解析

MCP 工作原理主要聚焦在 3 个方面，动态上下文窗口管理、上下文嵌入与压缩、有状态工作流。

1. 上下文窗口管理：MCP 采用动态上下文窗口，该窗口会随着每次交互不断扩展，存储以下信息：

1. 用户偏好（例如语言、语气）。

2. 对话历史（之前的查询 / 回复）。

3. 环境数据（如设备类型、地理位置）。

2. 上下文嵌入与压缩：为避免数据过载，MCP 将非关键数据压缩成嵌入形式（例如把 10 条消息的聊天记录总结为一个意图向量），同时保留关键细节。

3. 有状态工作流：MCP 支持多步骤工作流，智能体能够：

1. 记住过往操作（例如 “用户已上传其身份证”）。

2. 调整策略（例如若用户离线，则从电子邮件切换为短信）。

3. 依据反馈自我修正（例如 “用户不喜欢选项 A；优先考虑选项 B”） 。

下面这段代码定义了一个旅行智能体类 TravelAgent，它可以根据用户的历史交互信息和偏好来为用户预订航班。如果用户偏好经济舱，则在搜索航班时会考虑预算因素；否则，正常搜索航班。

# Hypothetical MCP stateful workflow (from official docs)class TravelAgent(MCPAgent):def __init__(self, user_id):self.context = load_context(user_id)# Load past interactionsself.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)

02 为何不直接让大模型访问应用程序 API ？

关于模型上下文协议（MCP），一个常见的问题是：“为什么我们需要一个定制协议呢？大语言模型难道不能自行学会使用应用程序编程接口（API）吗？”

从理论上讲，或许是可以的。大多数公开的 API 都配有说明其功能的文档。人们可以将这些文档提供给大语言模型，使其能够推导出实现目标所需的步骤。

然而在实际应用中，这种方法往往效率低下。作为致力于提升用户体验的开发者，必须优先考虑速度和响应性。

以一种易于大语言模型使用的格式为其提供工具，我们能够显著减少延迟，并简化整个流程。这种方式确保了用户能够获得更流畅的交互体验以及更快速的结果。

03 MCP vs 传统 API 调用

为何会出现 MCP 呢？

传统的应用程序编程接口（API）就如同快照，适用于静态任务。而模型上下文协议（MCP）则像是视频，能够完整捕捉用户意图的来龙去脉。

关于 MCP，我经常遇到这样一个问题：“它与工具调用有何不同？”

工具调用指的是大语言模型（LLM）调用函数以在现实世界中执行任务的机制。在这种机制下，大语言模型与一个工具执行器协同工作，由工具执行器调用指定工具并返回结果。其典型流程如下：

1. 描述要调用的工具

2. 发送结果

3. 大语言模型

4. 工具执行器

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

相比之下，MCP 框架允许大语言模型从一个独立的进程中访问工具，这个进程既可以在本地，也可以托管在远程服务器上。其结构如下：

1. MCP 服务器

2. MCP 客户端

3. 描述要调用的工具

4. 调用工具

5. 发送结果

6. 返回结果

7. 大语言模型

8. MCP 协议

9. 工具执行器

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

04 为什么 MCP 对智能体具有革命性意义

智能体框架要求 AI 不仅能做出响应，还能自主行动。MCP 的重要性体现在以下几个方面：

1. 实现真正的自主性，如今智能体能够：

1. 基于历史数据做出决策（例如，医疗智能体能调出患者的过敏清单）。

2. 无需人工干预即可串联任务（例如，完成 “研究→起草→编辑→发布” 博客文章的流程）。

2. 协同智能，MCP 允许智能体与以下对象共享上下文信息：

1. 其他智能体（例如，客服机器人将问题转交给人工客服）。

2. 外部工具（例如，将实时股票数据纳入金融顾问的回复中）。

3. 道德规范保障

1. 可审计性：完整的上下文历史记录有助于追溯有偏差或不准确的输出。

2. 隐私保护：敏感数据（如医疗记录）会被隔离存储。

如果没有 MCP，智能体就会缺乏连续性，就像厨师在烹饪过程中忘记菜谱步骤一样。

4. 实现长期自主性

1. 持久记忆：智能体能够记住用户偏好

2. 目标串联：执行多步骤任务（例如，“研究→谈判→预订商务旅行”）。

05 MCP 中的连接生命周期

MCP 中的连接生命周期对于管理客户端和服务器之间交互的状态和转换至关重要，可确保整个过程中通信的稳定和功能的正常运行。

MCP 中这种结构化的组件处理方式，为高效通信和集成提供了清晰的框架，有助于大模型（LLM）应用蓬勃发展。

1. 初始化：在初始化阶段，服务器和客户端之间会执行以下步骤：

1. 客户端发送包含协议版本及其功能的初始化请求。

2. 服务器回复自身的协议版本和功能。

3. 客户端发送初始化通知，确认连接成功建立。

4. 此时连接已准备就绪，可以开始正常的消息交换。
   

2. 消息交换：初始化阶段结束后，MCP 支持以下通信模式：

1. 请求 - 响应：客户端或服务器均可发送请求，另一方会做出响应。

2. 通知：任何一方都可以发送单向消息，无需等待回复。

3. 终止：任何一方都可以终止连接，有以下几种情况：

1. 正常关闭：通过 method.close () 方法实现。

2. 传输断开：通信通道丢失时发生。

3. 错误情况：遇到错误也可能导致连接终止。

错误处理

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

下面这段代码定义了一个枚举类型 ErrorCode，即是一组标准的 JSON - RPC 错误代码，在 MCP 系统里遇到相应问题时，就可以使用这些错误代码来准确地表示错误类型，进而方便进行错误处理和调试。

enum ErrorCode {// Standard JSON-RPC error codesParseError = -32700,InvalidRequest = -32600,MethodNotFound = -32601,InvalidParams = -32602,InternalError = -32603}

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

误差传播：

错误通过以下方式进行传达：

1. 错误响应：为响应有问题的请求而返回。

2. Error Events：在传输上触发以通知错误。

3. 协议级错误处理程序：在 MCP 级别管理错误。

这种结构化的生命周期可确保客户端和服务器之间可靠而高效的通信，同时在出现错误时妥善处理错误。

06 代码实现

下面我们通过比较有代表性的案例来看看，该案例实现展示了一种复杂的多服务器设置，该设置通过不同传输协议处理不同类型工作。

以下是该系统的直观表示形式图：

安装依赖项：通过 pip install mcp httpx langchain langchain-core langchai-community langchain-groq langchain-ollama langchain_mcp_adapters 命令安装所需的依赖包，并在 .env 文件中设置 groq api key。

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

import osfrom dotenv import load_dotenvload_dotenv()

3.1 数学服务器（math_server.py）：使用 mcp.server.fastmcp 模块中的 FastMCP 类创建一个名为 “Math” 的服务器，并定义了两个工具函数 add 和 multiply 分别用于两个整数的加法和乘法运算，最后通过 mcp.run(transport="stdio") 运行服务器。

# math_server.pyfrom 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")

from typing import Anyimport httpxfrom mcp.server.fastmcp import FastMCP# Initialize FastMCP servermcp = FastMCP("weather")# ConstantsNWS_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 Nonedef 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']}"""        forecasts.append(forecast)    return "\n---\n".join(forecasts)if __name__ == "__main__":    # Initialize and run the server    mcp.run(transport='sse')

4. 创建客户端（langchain_mcp_multiserver.py）：

加载环境变量，初始化 ChatOllama 模型。定义服务器参数 server_params，创建一个 MultiServerMCPClient 实例，其中配置了 “weather” 和 “math” 两个服务器的连接信息。定义异步函数 run_app，在函数中使用 create_react_agent 创建代理，调用 agent.ainvoke 处理用户问题并获取响应，最后打印响应内容。在 if __name__ == "__main__" 代码块中设置用户问题并运行 run_app 函数。

import asynciofrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientfrom IPython.display import display, Markdownfrom langchain_core.messages import HumanMessage, ToolMessage, AIMessagefrom langchain_mcp_adapters.tools import load_mcp_toolsfrom langgraph.prebuilt import create_react_agentfrom langchain_mcp_adapters.client import MultiServerMCPClientfrom langgraph.prebuilt import create_react_agent#from langchain_azure_ai.chat_models import AzureAIChatCompletionsModelfrom langchain_groq import ChatGroqfrom langchain_ollama import ChatOllamafrom langchain_core.prompts import PromptTemplatefrom dotenv import load_dotenv
load_dotenv()
#model = ChatGroq(model="llama-3.3-70b-versatile",temperature=0.5)model = ChatOllama(model="llama3.2:1b",temperature=0.0,max_new_tokens=500)server_params = StdioServerParameters(    command="python",    # Make sure to update to the full absolute path to your math_server.py file    args=["weather.py"],)async def run_app(user_question):            async with MultiServerMCPClient(        {            "weather": {                "url": "http://localhost:8000/sse",                "transport": "sse",            },            "math": {            "command": "python",            # Make sure to update to the full absolute path to your math_server.py file            "args": ["math_server.py"],            "transport": "stdio",            },        }    ) as client:        agent = create_react_agent(model, client.get_tools())        agent_response = await agent.ainvoke({"messages": user_question})        print(agent_response['messages'][-1].content)        # # Stream the response chunks        # async for chunk in agent.astream({"messages": user_question}):        #     # Extract the message content from the AddableUpdatesDict structure        #     if 'agent' in chunk and 'messages' in chunk['agent']:        #         for message in chunk['agent']['messages']:        #             if isinstance(message, AIMessage):        #                 # Handle different content formats        #                 if isinstance(message.content, list):        #                     # For structured content with text and tool use        #                     for item in message.content:        #                         if isinstance(item, dict) and 'text' in item:        #                             print(f"**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].contentif __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

The result of (3 + 5) is 8, and 8 x 12 is 96.

It appears you've provided a list of weather alerts from the National Weather Service (NWS) for various regions in New York State, Vermont, and parts of Massachusetts.
Here's a breakdown of what each alert is saying:**Flooding Alerts*** The NWS has issued several flood watches across New York State, including:        + Northern St. Lawrence; Northern Franklin; Eastern Clinton; Southeastern St. Lawrence; Southern Franklin; Western Clinton; Western Essex; Southwestern St. Lawrence; Grand Isle; Western Franklin; Orleans; Essex; Western Chittenden; Lamoille; Caledonia; Washington; Western Addison; Orange; Western Rutland; Eastern Franklin; Eastern Chittenden; Eastern Addison; Eastern Rutland; Western Windsor; Eastern Windsor        + Northern Herkimer; Hamilton; Southern Herkimer; Southern Fulton; Montgomery; Northern Saratoga; Northern Warren; Northern Washington; Northern Fulton; Southeast Warren; Southern Washington; Bennington; Western Windham; Eastern Windham* The NWS has also issued a flood watch for parts of Vermont, including:        + Northern New York and northern and central Vermont**Ice Jam Alerts*** The NWS has warned about the possibility of ice jams in several areas, including:        + Bennington; Western Windham; Eastern Windham        + Southern Vermont, Bennington and Windham Counties        + Central New York, Herkimer County        + Northern New York, Hamilton, Montgomery, Fulton, Herkimer, Warren, Washington Counties**Other Alerts*** The NWS has issued several warnings about heavy rainfall and snowmelt leading to minor river flooding.* There are also alerts for isolated ice jams that could further increase the flood risk.It's essential to stay informed about weather conditions in your area and follow the instructions of local authorities. If you're planning outdoor activities, be prepared for changing weather conditions and take necessary precautions to stay safe.It appears you've provided a list of weather alerts from the National Weather Service (NWS) for various regions in New York State, Vermont, and parts of Massachusetts.Here's a breakdown of what each alert is saying:**Flooding Alerts*** The NWS has issued several flood watches across New York State, including:        + Northern St. Lawrence; Northern Franklin; Eastern Clinton; Southeastern St. Lawrence; Southern Franklin; Western Clinton; Western Essex; Southwestern St. Lawrence; Grand Isle; Western Franklin; Orleans; Essex; Western Chittenden; Lamoille; Caledonia; Washington; Western Addison; Orange; Western Rutland; Eastern Franklin; Eastern Chittenden; Eastern Addison; Eastern Rutland; Western Windsor; Eastern Windsor        + Northern Herkimer; Hamilton; Southern Herkimer; Southern Fulton; Montgomery; Northern Saratoga; Northern Warren; Northern Washington; Northern Fulton; Southeast Warren; Southern Washington; Bennington; Western Windham; Eastern Windham* The NWS has also issued a flood watch for parts of Vermont, including:        + Northern New York and northern and central Vermont**Ice Jam Alerts*** The NWS has warned about the possibility of ice jams in several areas, including:        + Bennington; Western Windham; Eastern Windham        + Southern Vermont, Bennington and Windham Counties        + Central New York, Herkimer County        + Northern New York, Hamilton, Montgomery, Fulton, Herkimer, Warren, Washington Counties**Other Alerts*** The NWS has issued several warnings about heavy rainfall and snowmelt leading to minor river flooding.* There are also alerts for isolated ice jams that could further increase the flood risk.
It's essential to stay informed about weather conditions in your area and follow the instructions of local authorities. If you're planning outdoor activities, be prepared for changing weather conditions and take necessary precautions to stay safe.

然后，MCP 客户端能够根据问题连接到相应服务器了。

如果有分析不对的地方还请多多指教，也欢迎对 MCP 感兴趣的朋友私我?进 MCP&Agent 交流群（备注MCP）。

Last but not least

上面 MCP 的实现过程，展示了其用于构建复杂 AI 应用程序的强大、可扩展且灵活的架构。此模块化设计和对多种传输协议的支持使 AI 智能体有了更好的选择，特别是用于多模态 AI 应用、复杂的工作流编排、分布式 AI 系统、实时数据处理应用程序等等。

对于智能体，主动行动才是最重要的，而不是被动行动。