[智能体设计模式] 第10章：模型上下文协议（MCP）

为了让大语言模型（LLM）能够有效地作为智能体（Agent）工作，其能力必须超越多模态生成，能够与外部环境交互，包括访问实时数据、调用外部软件、执行具体操作任务。模型上下文协议（MCP）正是为此而设计，它为 LLM 与外部资源的对接提供了标准化接口，是实现一致性和可预测集成的关键机制。

MCP 模式概述

可以将 MCP 想象成一个通用适配器，让任何 LLM 都能无缝连接到任何外部系统、数据库或工具，无需为每种组合单独开发集成。MCP 是一项开放标准，旨在规范 Gemini、OpenAI GPT、Mixtral、Claude 等 LLM 与外部应用、数据源和工具的通信方式。它就像一个通用连接机制，简化了 LLM 获取上下文、执行操作、与各种系统交互的流程。

MCP 采用客户端-服务器架构。MCP 服务器负责暴露数据（资源）、交互模板（即 Prompt）和可执行功能（工具），而 MCP 客户端则负责消费这些能力，客户端可以是 LLM 宿主应用或智能体本身（本文中为 LangChain 智能体）。这种标准化方式极大降低了 LLM 集成到多样化业务环境的复杂度。

需要注意的是，MCP 本质上是一种“智能体接口”契约，其效果高度依赖于底层 API 的设计。如果开发者只是简单地将传统 API 包装为 MCP 接口，而不做优化，智能体的表现可能很差。例如，某工单系统 API 只能逐条获取工单详情，智能体要汇总高优先级工单时就会很慢且不准确。要真正发挥智能体优势，底层 API 应支持确定性特性，如过滤和排序，帮助智能体高效工作。智能体并不能神奇地替代确定性流程，往往需要更强的确定性支持。

此外，MCP 可以包装任何 API，但如果 API 的输入输出格式智能体无法理解，依然无效。例如，文档存储 API 只返回 PDF 文件，智能体无法解析 PDF 内容，这样的 MCP 服务就没有实际意义。更好的做法是先开发一个能返回文本（如 Markdown）的 API，让智能体能直接读取和处理。这说明开发者不仅要关注连接方式，更要关注数据本身的可用性，确保真正的兼容性。

MCP 与工具函数调用的区别

模型上下文协议（MCP）与工具函数调用是 LLM 扩展外部能力的两种机制。二者都能让 LLM 执行文本生成之外的操作，但在抽象层次和实现方式上有明显区别。

工具函数调用是 LLM 直接向某个预定义工具或函数发起请求（“工具”和“函数”在此语境下可互换）。这种方式是一对一通信，LLM 根据用户意图格式化请求，应用代码执行后返回结果。不同 LLM 厂商实现方式各异，通常是专有的。

而 MCP 则是一个标准化接口，让 LLM 能够发现、通信并调用外部能力。它是开放协议，支持 LLM 与各种工具和系统的交互，目标是建立一个任何合规工具都能被任何合规 LLM 访问的生态系统，促进互操作性、可组合性和复用性。采用联邦模型后，能显著提升系统间的协同和资产价值。只需将传统服务包装为 MCP 接口，无需重写底层系统，就能将其纳入现代智能体生态，实现敏捷复用。

以下是 MCP 与工具函数调用的核心区别：

特性	工具函数调用	模型上下文协议（MCP）
标准化	专有、厂商定制，格式和实现各异	开放标准协议，促进 LLM 与工具间互操作
范围	LLM 直接请求某个预定义函数	LLM 与外部工具发现和通信的通用框架
架构	LLM 与应用工具逻辑一对一交互	客户端-服务器架构，LLM 应用可连接多个 MCP 服务器
发现机制	需显式告知 LLM 可用工具	支持动态发现，客户端可查询服务器能力
复用性	工具集成与应用和 LLM 高度耦合	支持开发可复用、独立的 MCP 服务器，任何应用可访问

工具函数调用就像给 AI 配一套专用工具（如特定扳手和螺丝刀），适合固定任务场景；而 MCP 则像通用电源插座系统，不直接提供工具，但允许任何合规工具接入，打造动态、可扩展的智能体工作坊。

简言之，函数调用适合简单场景，MCP 则是复杂、互联 AI 系统不可或缺的标准化通信框架。

MCP 的更多考量

MCP 虽强大，但实际应用需综合考虑以下关键因素：

• 工具、资源与 Prompt 的区别：资源是静态数据（如 PDF、数据库记录），工具是可执行功能（如发邮件、API 查询），Prompt 是引导 LLM 与资源或工具交互的模板，确保结构化和高效互动。
• 可发现性：MCP 客户端可动态查询服务器能力，实现“即时发现”，智能体无需重启即可适应新功能。
• 安全性：任何协议暴露工具和数据都需强安全措施。MCP 实现必须支持认证和授权，控制客户端访问权限和操作范围。
• 实现复杂度：MCP 虽为开放标准，但实现可能较复杂。部分厂商（如 Anthropic、FastMCP）已推出 SDK，简化开发流程。
• 错误处理：协议需定义错误（如工具执行失败、服务器不可用、请求无效）如何反馈给 LLM，便于智能体理解并尝试替代方案。
• 本地与远程服务器：MCP 服务器可部署在本地或远程。本地适合敏感数据和高性能场景，远程则便于组织共享和扩展。
• 按需与批量处理：MCP 支持实时交互和批量处理，适用于对话型智能体和数据分析流水线等不同场景。
• 传输机制：本地通信采用 JSON-RPC over STDIO，高效进程间交互；远程则用 Streamable HTTP 和 SSE，支持持久高效的客户端-服务器通信。

MCP 采用客户端-服务器模型，标准化信息流。理解各组件交互是实现高级智能体行为的关键：

1. LLM：核心 Agent，处理用户请求、制定计划、决定何时访问外部信息或执行操作。
2. MCP 客户端：LLM 的应用或包装层（本文中为 LangChain 智能体），将 LLM 意图转化为 MCP 标准请求，负责发现、连接和通信。
3. MCP 服务器：外部世界的入口，向授权客户端暴露工具、资源和 Prompt，通常负责某一领域（如数据库、邮件服务、API）。
4. 第三方服务：实际的外部工具、应用或数据源，由 MCP 服务器管理和暴露，是最终执行操作的终点（如数据库查询、SaaS 平台、天气 API）。

交互流程如下：

1. 发现：MCP 客户端（LangChain 智能体）代表 LLM 查询服务器能力，服务器返回工具、资源和 Prompt 清单。
2. 请求构造：LLM 决定使用某工具（如发邮件），构造请求并指定参数（收件人、主题、正文）。
3. 客户端通信：MCP 客户端将请求按标准格式发送至 MCP 服务器。
4. 服务器执行：MCP 服务器认证客户端、校验请求，调用底层软件执行操作（如邮件 API 的 send 函数）。
5. 响应与上下文更新：服务器返回标准化响应（如邮件发送确认 ID），客户端将结果反馈给 LLM，更新上下文，智能体继续后续任务。

实践应用与场景

MCP 极大拓展了 AI/LLM 能力，常见九大应用场景：

• 数据库集成：智能体可通过 MCP 无缝访问结构化数据库，如用 MCP 数据库工具箱查询 Google BigQuery，实时获取信息、生成报告或更新记录，全部由自然语言驱动。
• 生成式媒体编排：智能体可集成高级生成媒体服务，如通过 MCP 工具调用 Google Imagen 生成图片、Veo 生成视频、Chirp 3 HD 生成语音、Lyria 生成音乐，实现 AI 应用中的动态内容创作。
• 外部 API 交互：MCP 为 LLM 调用外部 API 提供标准化方式，智能体可获取实时天气、股票价格、发送邮件、对接 CRM 系统，能力远超基础模型。
• 推理型信息抽取：利用 LLM 强推理能力，MCP 支持智能体按需抽取信息，超越传统检索工具。智能体可分析文本，精准提取回答复杂问题的关键句段。
• 自定义工具开发：开发者可用 FastMCP 等框架快速开发自定义工具，并通过 MCP 服务器暴露，无需修改 LLM 即可让智能体访问专有功能。
• 标准化 LLM-应用通信：MCP 为 LLM 与应用间通信提供一致层，降低集成成本，促进不同厂商和宿主应用间互操作，简化复杂智能体系统开发。
• 复杂流程编排：智能体可组合多种 MCP 工具和数据源，实现多步骤复杂流程，如获取客户数据、生成营销图片、撰写邮件并发送，全部自动化完成。
• 物联网设备控制：智能体可通过 MCP 控制 IoT 设备，如智能家居、工业传感器、机器人，实现自然语言驱动的自动化。
• 金融服务自动化：在金融领域，智能体可通过 MCP 对接数据源、交易平台、合规系统，实现市场分析、自动交易、个性化建议和合规报告，确保安全和标准化通信。

简言之，MCP 让智能体能访问数据库、API 和网页等实时信息，也能执行发邮件、更新记录、控制设备等复杂任务，并支持 AI 应用中的媒体生成工具集成。

LangChain 实操代码示例

本节介绍如何基于 LangChain 框架连接本地 MCP 服务器，实现 LangChain 智能体与本地文件系统、FastMCP 服务器的交互。

核心前提

1. 依赖安装：

# 基础依赖（LangChain 核心 + Gemini 模型）
pip install langchain langchain-google-genai python-dotenv requests# MCP 相关依赖（文件系统 MCP 服务器 + FastMCP）
npm install -g @modelcontextprotocol/server-filesystem  # 全局安装文件系统 MCP 服务器
pip install fastmcp  # FastMCP 框架

2. 环境配置：
   创建 .env 文件，配置 Gemini API 密钥：

GOOGLE_API_KEY=你的 Gemini API 密钥（从 Google AI Studio 获取）

场景 1：LangChain 智能体 + MCP 文件系统交互

原 ADK 功能：通过 MCP 文件系统服务器，让智能体读写本地文件。LangChain 实现保持功能一致，通过自定义 MCP 工具封装服务器调用。

步骤 1：启动 MCP 文件系统服务器

先启动 MCP 文件服务器：

# 替换为你的目标文件夹路径（如 ./mcp_managed_files）
npx @modelcontextprotocol/server-filesystem /path/to/your/folder

服务器默认监听 http://localhost:8080（可通过 --port 参数修改）。

步骤 2：LangChain 智能体配置

创建 langchain_mcp_file_agent.py，实现文件管理智能体：

import os
import requests
from dotenv import load_dotenv
from langchain.agents import AgentType, initialize_agent, Tool
from langchain_google_genai import ChatGoogleGenerativeAI# 加载环境变量（Gemini API 密钥）
load_dotenv()# --------------------------
# 1. 配置 MCP 文件服务器地址和目标目录
# --------------------------
MCP_FILE_SERVER_URL = "http://localhost:8080"  # MCP 服务器默认地址
# 获取脚本同级目录下'mcp_managed_files'文件夹的绝对路径
TARGET_FOLDER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mcp_managed_files")
os.makedirs(TARGET_FOLDER_PATH, exist_ok=True)  # 确保文件夹存在# --------------------------
# 2. 自定义 MCP 文件操作工具（封装 MCP 服务器 API）
# MCP 协议核心接口：/tools/list_directory、/tools/read_file、/tools/write_file
# --------------------------
def call_mcp_tool(tool_name: str, params: dict) -> str:"""通用 MCP 工具调用函数（发送 HTTP 请求到 MCP 服务器）"""try:response = requests.post(f"{MCP_FILE_SERVER_URL}/tools/{tool_name}",json={"parameters": params},timeout=10)response.raise_for_status()  # 抛出 HTTP 错误return response.json()["result"]except Exception as e:return f"MCP 工具调用失败：{str(e)}"# 工具 1：列出目录内容
def list_directory(path: str = "") -> str:"""列出 MCP 服务器管理目录下的文件/文件夹（path 为相对路径，默认根目录）"""full_path = os.path.join(TARGET_FOLDER_PATH, path) if path else TARGET_FOLDER_PATHreturn call_mcp_tool("list_directory", {"path": full_path})# 工具 2：读取文件
def read_file(file_path: str) -> str:"""读取 MCP 服务器管理目录下的文件（file_path 为相对路径）"""full_path = os.path.join(TARGET_FOLDER_PATH, file_path)return call_mcp_tool("read_file", {"path": full_path})# 工具 3：写入文件
def write_file(file_path: str, content: str) -> str:"""向 MCP 服务器管理目录下的文件写入内容（file_path 为相对路径）"""full_path = os.path.join(TARGET_FOLDER_PATH, file_path)return call_mcp_tool("write_file", {"path": full_path, "content": content})# --------------------------
# 3. 定义 LangChain 工具集
# --------------------------
file_tools = [Tool(name="list_directory",func=list_directory,description=f"列出目标目录的内容，目标目录为：{TARGET_FOLDER_PATH}。无需传入参数时默认列出根目录，支持传入相对路径查看子目录。"),Tool(name="read_file",func=read_file,description=f"读取目标目录下的文件内容，需传入文件相对路径（例如：sample.txt、docs/another.md），目标目录为：{TARGET_FOLDER_PATH}。"),Tool(name="write_file",func=write_file,description=f"向目标目录下的文件写入内容，需传入两个参数：file_path（文件相对路径）和 content（要写入的内容），目标目录为：{TARGET_FOLDER_PATH}。")
]# --------------------------
# 4. 初始化 LangChain 智能体
# --------------------------
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0.3
)# 初始化 Agent（采用结构化聊天型 Agent，优化工具调用逻辑）
file_agent = initialize_agent(tools=file_tools,llm=llm,agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,  # 打印思考过程（调试用）agent_kwargs={"system_message": ("你是文件管理助手，负责管理目标目录：{TARGET_FOLDER_PATH}。""可使用提供的工具列出目录、读取文件、写入文件。""请根据用户需求选择合适的工具，参数格式要正确。").format(TARGET_FOLDER_PATH=TARGET_FOLDER_PATH)}
)# --------------------------
# 5. 测试智能体
# --------------------------
if __name__ == "__main__":# 测试示例print("=== 测试 1：显示文件夹内容 ===")print(file_agent.run("显示该文件夹内容。"))print("\n=== 测试 2：读取 sample.txt 文件 ===")print(file_agent.run("读取 sample.txt 文件。"))print("\n=== 测试 3：写入 another_file.md ===")print(file_agent.run("向 another_file.md 写入内容：'这是 LangChain 替代 ADK 的 MCP 文件操作示例'"))

场景 2：LangChain 智能体 + FastMCP 服务器交互

步骤 1：创建并启动 FastMCP 服务器

创建 fastmcp_server.py，实现 greet 工具服务器：

# fastmcp_server.py
from fastmcp import FastMCPmcp_server = FastMCP()@mcp_server.tool
def greet(name: str) -> str:"""生成个性化问候语。Args:name: 要问候的人名。Returns:问候语字符串。"""return f"你好，{name}！很高兴认识你。"if __name__ == "__main__":mcp_server.run(transport="http",host="127.0.0.1",port=8000  # 监听 8000 端口)

启动服务器：

python fastmcp_server.py

步骤 2：LangChain 客户端智能体配置

创建 langchain_fastmcp_agent.py，实现连接 FastMCP 服务器的智能体：

import requests
from dotenv import load_dotenv
from langchain.agents import AgentType, initialize_agent, Tool
from langchain_google_genai import ChatGoogleGenerativeAI# 加载环境变量（Gemini API 密钥）
load_dotenv()# --------------------------
# 1. 配置 FastMCP 服务器地址
# --------------------------
FASTMCP_SERVER_URL = "http://localhost:8000"  # FastMCP 服务器地址# --------------------------
# 2. 自定义 FastMCP 工具调用（封装 greet 工具）
# --------------------------
def call_fastmcp_greet(name: str) -> str:"""调用 FastMCP 服务器的 greet 工具，生成个性化问候语"""try:response = requests.post(f"{FASTMCP_SERVER_URL}/tools/greet",json={"parameters": {"name": name}},timeout=5)response.raise_for_status()  # 抛出 HTTP 错误return response.json()["result"]except Exception as e:return f"FastMCP 工具调用失败：{str(e)}"# --------------------------
# 3. 定义 LangChain 工具集（仅暴露 greet 工具）
# --------------------------
fastmcp_tools = [Tool(name="greet",func=call_fastmcp_greet,description="生成个性化问候语，必须传入参数：name（要问候的人名）。")
]# --------------------------
# 4. 初始化 LangChain 智能体
# --------------------------
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0.3
)fastmcp_agent = initialize_agent(tools=fastmcp_tools,llm=llm,agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,verbose=True,agent_kwargs={"system_message": "你是一个友好的助手，只能通过 'greet' 工具向人问好。请根据用户需求提取人名参数，调用工具生成问候语。"}
)# --------------------------
# 5. 测试智能体
# --------------------------
if __name__ == "__main__":print("=== 测试：向 John Doe 问好 ===")print(fastmcp_agent.run("向 John Doe 问好"))print("\n=== 测试：向 Alice 问好 ===")print(fastmcp_agent.run("请问候一下 Alice"))