加载中

现在我们的 MCP 服务器已经可以正常工作了,是时候构建客户端了。客户端是让我们的应用程序能够与 MCP 服务器通信并访问其功能的部分。

理解客户端架构

在深入代码之前,让我们先澄清一下关于 MCP 项目的一个重要问题。通常情况下,您会实现 MCP 客户端或 MCP 服务器中的一个——而不是两者都实现。在本项目中我们两者都构建,只是为了让您看看它们是如何协同工作的。

我们服务器的示意图,其中包含 MCP 客户端,该客户端以双向方式连接到一个独立的 MCP 服务器。

MCP 客户端由两个主要组件组成:

  • MCP Client - 我们创建的一个自定义类,用于简化会话的使用
  • Client Session - 与服务器的实际连接(属于 MCP Python SDK 的一部分)

mcp_client.py 的示意图,其中包含 MCP Client 和 Client Session,后者连接到一个独立的 MCP 服务器。

客户端会话在我们使用完毕后需要进行资源清理,这就是我们将其封装在自定义类中的原因。这样可以自动处理连接管理和清理工作。

客户端如何融入我们的应用程序

还记得我们的应用程序流程吗?我们的 CLI 代码需要以两种关键方式与 Claude 交互:

应用程序流程的时序图,突出显示了我们的 CLI 代码在何处使用客户端来列出工具和调用工具。

客户端通过将服务器的功能暴露给我们的代码库,来实现这两种交互方式。

实现核心客户端函数

我们需要实现两个基本函数:list_tools 和 call_tool。

列出工具函数

该函数从服务器获取所有可用的工具:

python
async def list_tools(self) -> list[types.Tool]:
    result = await self.session().list_tools()
    return result.tools

这很简单——我们访问我们的会话(即与 MCP 服务器的连接),调用内置的 list_tools 函数,并从结果中返回工具列表。

调用工具函数

该函数在服务器上执行指定的工具:

python
async def call_tool(
    self, tool_name: str, tool_input: dict
) -> types.CallToolResult | None:
    return await self.session().call_tool(tool_name, tool_input)

我们将工具名称和输入参数(由 Claude 提供)传递给服务器,并返回结果。

测试客户端

为了验证我们的实现是否有效,我们可以直接对其进行测试。客户端文件中包含一个测试工具,它会连接到 MCP 服务器并对其运行命令。让它调用 list_tools() 并打印结果:

python
async with MCPClient(
    command="uv", args=["run", "mcp_server.py"]
) as client:
    result = await client.list_tools()
    print(result)

运行 uv run mcp_client.py 应该会返回一个可用工具的列表,以及它们的描述和输入模式(input schema)。您应该能看到我们在服务器中定义的工具,例如 read_doc_contents 和 edit_document。

端到端测试

既然客户端和服务器都已正常工作,我们就可以测试完整的流程了。运行我们的主应用程序,并询问 Claude:"report.pdf 文档的内容是什么?"应该会发生以下情况:

  1. 将可用工具列表发送给 Claude
  2. Claude 决定使用 read_doc_contents 工具
  3. 我们的客户端在服务器上调用该工具
  4. 服务器返回文档内容
  5. Claude 给出包含该信息的回复

客户端充当您的应用程序代码与 MCP 服务器之间的桥梁,使您能够轻松访问服务器的功能,而无需直接处理底层的连接细节。