MCP客户端智能体的开发实践

推荐语

揭秘MCP客户端智能体的开发全流程，掌握如何以编程方式高效调用MCP服务器。

核心内容：
1. MCP架构的三大核心组件解析
2. 自定义MCP客户端的实现方法与优势
3. 多服务器协同工作的完整交互流程

杨芳贤

53AI创始人/腾讯云(TVP)最具价值专家

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 并初始化它。