1 引言
本文旨在深入剖析模型上下文协议（MCP）架构及其客户端流程，并实现一个 MCP 客户端智能体。我们将明确当向由大语言模型（LLM）支持的 MCP 提出请求时，幕后究竟发生了什么，以期让整个过程更加清晰易懂。
目前，关于构建 MCP 服务器的文章已有很多，例如 MCP 官方网站 上的教程，详细介绍了如何构建一个简单的 MCP 天气服务器并将其连接到主机（如 Claude Desktop）。不过，本文将专注于实现一个能够以编程方式连接到 MCP 服务器的 MCP 客户端智能体。
2 高级MCP架构
2.1 MCP组件
主机（Host）：这是用户直接交互的界面，如 Claude Desktop 或 Cursor 等 AI 代码编辑器，它充当主界面和系统管理器的角色。
客户端（Client）：作为主机和 MCP 服务器之间的中介，负责维护连接，处理通信协议以及数据流的传输。
服务器（Server）：通过标准化接口为 AI 模型提供特定功能、数据源和工具的组件，例如天气预报、文件读取等服务。
接下来，我们将深入探讨 MCP 客户端智能体的实现。
3 MCP客户端智能体
3.1 自定义MCP客户端：以编程方式调用MCP服务器
在目前常见的用例中，MCP 主要应用于 AI 驱动的集成开发环境（IDE）。在这种场景下，用户在 IDE 中配置 MCP 服务器，并通过聊天界面与它们进行交互，此时聊天界面就充当了 MCP 客户端或主机的角色。
然而，如果我们希望从自己的服务中以编程方式调用 MCP 服务器，又该如何实现呢？这正是 MCP 的核心优势所在。它为 LLM 提供了一种标准化的方式来获取上下文和工具。开发者无需为每个外部 API、资源或文件编写自定义代码进行集成，而是可以专注于打包正确的上下文和功能，然后将它们交给 LLM 进行推理和处理，从而极大地提高了开发效率和系统的可扩展性。
4 具有多个MCP服务器的MCP客户端智能体工作流程
上图展示了 MCP 自定义客户端（或 AI 智能体）如何通过多个 MCP 服务器处理用户请求。以下是该交互流程的详细步骤分解：
步骤1：用户发起请求
用户通过 IDE、浏览器或终端提交请求。 自定义 MCP 客户端（或 AI 智能体界面）接收用户的查询。
步骤2：MCP客户端与服务器连接
MCP 客户端与一个或多个 MCP 服务器建立连接，并从这些服务器请求可用的工具。 每个连接的服务器返回其支持的工具和函数列表。
步骤3：AI处理
用户查询和服务器返回的工具列表一起发送到 LLM（例如 OpenAI）。 LLM 分析用户请求，并根据工具列表选择合适的工具及其输入参数，然后将响应发送回 MCP 客户端。
步骤4：函数执行
MCP 客户端根据 LLM 的建议，使用指定的参数调用相应 MCP 服务器中的工具。 MCP 服务器接收工具调用请求，并根据请求内容，调用其内部的相应工具进行处理。请注意，为避免 LLM 产生幻觉和非确定性响应，应确保不同 MCP 服务器中的工具名称具有唯一性。 MCP 服务器可以与数据库、外部 API 或文件系统进行交互，以完成请求的处理。
步骤5：（可选）使用LLM改进响应
MCP 服务器将工具执行的结果返回给 MCP 客户端。 （可选）MCP 客户端可以将该结果转发给 LLM 进行进一步优化，例如将技术性响应转换为自然语言描述，或创建更简洁的摘要。
步骤6：回复用户
最终处理后的响应通过客户端界面发送回用户。 用户收到对其原始查询的答复。
5 自定义MCP客户端实现/源代码
5.1 连接到MCP服务器
如前所述，MCP 客户端能够连接到多个 MCP 服务器。在自定义 MCP 客户端实现中，可以模拟这样的行为。
注意： 为了减少幻觉并确保结果的一致性，建议避免多个 MCP 服务器之间的工具名称冲突。
5.2 MCP服务器传输选项
MCP 服务器支持两种类型的传输机制：
- STDIO
：用于本地进程间通信。 - SSE
：用于基于 HTTP/WebSocket 的请求。
5.3 连接到STDIO传输
以下是连接到 STDIO 传输的示例代码：
asyncdefconnect_to_stdio_server(self, server_script_path: str):
"""
连接到 MCP 标准输入输出（stdio）服务器。
参数:
server_script_path (str): 服务器脚本的路径。
"""
is_python = server_script_path.endswith('.py')
is_js = server_script_path.endswith('.js')
ifnot (is_python or is_js):
raise ValueError("服务器脚本必须是 .py 或 .js 文件")
command = "python"if is_python else"node"
server_params = StdioServerParameters(
command=command,
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
print("已初始化 stdio...")
connect_to_stdio_server：此函数用于连接到 STDIO 传输的 MCP 服务器。它首先检查服务器脚本的文件扩展名，以确定使用 Python 还是 Node.js 来运行服务器脚本。然后，它创建一个 StdioServerParameters 对象，并使用该对象初始化 stdio_client。最后，它创建一个 ClientSession 并初始化它。
5.4 连接到SSE传输
以下是连接到 SSE 传输的示例代码：
asyncdefconnect_to_sse_server(self, server_url: str):
"""
连接到使用 SSE 传输的 MCP 服务器。
参数:
server_url (str): 服务器的 URL。
"""
# 存储上下文管理器以保持其处于活动状态
self._streams_context = sse_client(url=server_url)
streams = await self._streams_context.__aenter__()
self._session_context = ClientSession(*streams)
self.session: ClientSession = await self._session_context.__aenter__()
await self.session.initialize()
print("已初始化 SSE...")
connect_to_sse_server：此函数用于连接到使用 SSE 传输的 MCP 服务器。它创建一个 sse_client 对象，并使用该对象初始化 ClientSession。然后，它创建一个 ClientSession 并初始化它。
