1.构建一个MCP客户端 #

开始构建您自己的客户端，使其能够与所有MCP服务器集成。

2.🖥️ 系统要求 #

在开始之前，请确保您的系统满足以下要求：

• 操作系统: Mac 或 Windows 电脑
• Python: 已安装最新版Python
• 包管理器: 最新版本 uv 已安装

3.⚙️ 环境设置 #

首先，创建一个新的Python项目：

# 创建项目目录
uv init mcp-client
cd mcp-client

# 创建虚拟环境
uv venv

# 激活虚拟环境
# Windows:
.venv\Scripts\activate
# Unix或MacOS:
source .venv/bin/activate

# 安装必需的包
uv add mcp anthropic python-dotenv

# 删除样板文件
# Windows:
del main.py
# Unix或MacOS:
rm main.py

# 创建我们的主文件
touch client.py

4. API密钥配置 #

您需要一个来自Anthropic的API密钥，可以从Anthropic控制台获取。

4.1 创建环境变量文件 #

# 创建.env文件
touch .env

4.2 添加API密钥 #

在 .env 文件中添加您的密钥：

ANTHROPIC_API_KEY=<your key here>

4.3 配置Git忽略文件 #

# 添加.env到.gitignore
echo ".env" >> .gitignore

⚠️ 安全提醒: 确保您妥善保管您的 ANTHROPIC_API_KEY！

5.🏗️ 创建客户端 #

5.1 基础客户端结构 #

首先，让我们设置导入并创建基本的client类：

# 导入异步IO库
import asyncio
# 导入可选类型注解
from typing import Optional
# 导入异步上下文管理器堆栈
from contextlib import AsyncExitStack

# 从mcp库导入客户端会话和标准输入输出服务器参数
from mcp import ClientSession, StdioServerParameters
# 从mcp客户端模块导入stdio客户端工厂方法
from mcp.client.stdio import stdio_client

# 导入Anthropic官方SDK
from anthropic import Anthropic
# 导入用于加载环境变量的库
from dotenv import load_dotenv

# 加载.env文件中的环境变量
load_dotenv()  # 从.env加载环境变量

# 定义MCP客户端类
class MCPClient:
    # 构造方法
    def __init__(self):
        # 初始化会话对象为None
        self.session: Optional[ClientSession] = None
        # 创建异步退出堆栈，用于管理异步上下文
        self.exit_stack = AsyncExitStack()
        # 初始化Anthropic客户端对象
        self.anthropic = Anthropic()
    # 方法将在这里添加

5.2 服务器连接管理 #

接下来，我们将实现连接MCP服务器的方法：

    # 定义异步方法，用于连接到MCP服务器
    async def connect_to_server(self, server_script_path: str):
        """连接到MCP服务器

        Args:
            server_script_path: 服务器脚本路径 (.py 或 .js)
        """
        # 判断脚本路径是否以.py结尾，表示Python脚本
        is_python = server_script_path.endswith('.py')
        # 判断脚本路径是否以.js结尾，表示Node.js脚本
        is_js = server_script_path.endswith('.js')

        # 如果既不是.py也不是.js，抛出异常
        if not (is_python or is_js):
            raise ValueError("服务器脚本必须是.py或.js文件")

        # 根据脚本类型选择启动命令
        command = "python" if is_python else "node"
        # 构造StdioServerParameters对象，指定命令、参数和环境变量
        server_params = StdioServerParameters(
            command=command,
            args=[server_script_path],
            env=None
        )

        # 通过异步上下文管理器启动stdio客户端，获取传输通道
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
        # 解包传输通道，分别为读取和写入对象
        self.stdio, self.write = stdio_transport
        # 创建MCP客户端会话，并进入异步上下文
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

        # 初始化会话，完成握手
        await self.session.initialize()

        # 列出服务器可用的工具
        response = await self.session.list_tools()
        # 获取工具列表
        tools = response.tools
        # 打印可用工具名称
        print(f"\n已连接到服务器，可用工具:", [tool.name for tool in tools])

5.3 查询处理逻辑 #

现在让我们添加处理查询和工具调用的核心功能：

# 定义异步方法，处理用户查询，返回字符串结果
async def process_query(self, query: str) -> str:
    """使用Claude和可用工具处理查询"""
    # 构造初始消息列表，用户输入作为第一条消息
    messages = [
        {
            "role": "user",
            "content": query
        }
    ]

    # 异步获取服务器可用工具列表
    response = await self.session.list_tools()
    # 构建Claude API所需的工具描述列表
    available_tools = [{
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    } for tool in response.tools]

    # 首次调用Claude API，传入模型、最大token、消息和工具列表
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )

    # 用于收集最终输出文本
    final_text = []

    # 用于记录assistant消息内容
    assistant_message_content = []
    # 遍历Claude返回的内容
    for content in response.content:
        # 如果是文本内容，直接添加到输出和assistant内容
        if content.type == 'text':
            final_text.append(content.text)
            assistant_message_content.append(content)
        # 如果是工具调用请求
        elif content.type == 'tool_use':
            # 获取工具名称和参数
            tool_name = content.name
            tool_args = content.input

            # 异步调用MCP工具，获取结果
            result = await self.session.call_tool(tool_name, tool_args)
            # 记录工具调用信息到输出
            final_text.append(f"[调用工具 {tool_name} 参数 {tool_args}]")

            # assistant消息内容追加本次工具调用
            assistant_message_content.append(content)
            # 将assistant消息加入消息历史
            messages.append({
                "role": "assistant",
                "content": assistant_message_content
            })
            # 将工具调用结果作为user消息加入消息历史
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": content.id,
                        "content": result.content
                    }
                ]
            })

            # 再次调用Claude API，获取工具调用后的新回复
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )

            # 将新回复的文本内容添加到输出
            final_text.append(response.content[0].text)

    # 返回所有输出文本拼接后的字符串
    return "\n".join(final_text)

5.4 交互式聊天界面 #

现在我们将添加聊天循环和清理功能：

# 定义异步方法chat_loop，用于运行交互式聊天循环
async def chat_loop(self):
    # 打印客户端启动提示
    print("\nMCP客户端已启动!")
    # 提示用户输入查询或退出指令
    print("输入您的查询或输入'quit'退出。")

    # 进入无限循环，持续接收用户输入
    while True:
        try:
            # 获取用户输入并去除首尾空白字符
            query = input("\n查询: ").strip()

            # 如果用户输入'quit'（不区分大小写），则跳出循环
            if query.lower() == 'quit':
                break

            # 调用process_query方法处理用户输入，并等待异步结果
            response = await self.process_query(query)
            # 打印处理结果
            print("\n" + response)

        # 捕获并打印异常信息
        except Exception as e:
            print(f"\n错误: {str(e)}")

# 定义异步方法cleanup，用于清理资源
async def cleanup(self):
    # 关闭exit_stack，释放相关资源
    await self.exit_stack.aclose()

5.5 主入口点 #

最后，我们将添加主要的执行逻辑：

# 定义异步主函数
async def main():
    # 检查命令行参数数量是否小于2
    if len(sys.argv) < 2:
        # 如果参数不足，打印用法提示
        print("用法: python client.py <服务器脚本路径>")
        # 退出程序，返回状态码1
        sys.exit(1)

    # 创建MCPClient实例
    client = MCPClient()
    try:
        # 连接到指定的服务器脚本
        await client.connect_to_server(sys.argv[1])
        # 启动交互式聊天循环
        await client.chat_loop()
    finally:
        # 无论如何都要清理资源
        await client.cleanup()

# 判断当前模块是否为主程序入口
if __name__ == "__main__":
    # 导入sys模块
    import sys
    # 使用asyncio运行主函数
    asyncio.run(main())

您可以在这里找到完整的 client.py 文件。

6.🔧 关键组件详解 #

6.1 客户端初始化 #

• MCPClient类 初始化时包含会话管理和API客户端
• 使用 AsyncExitStack 妥善管理资源
• 配置Anthropic客户端以实现与Claude的交互

6.2 服务器连接 #

• 支持Python和Node.js服务器
• 验证服务器脚本类型
• 建立适当的沟通渠道
• 初始化会话并列出可用工具

6.3 查询处理 #

• 保持对话上下文
• 处理Claude的响应和工具调用
• 管理Claude与工具之间的消息流
• 将结果整合成一个连贯的响应

6.4 交互式界面 #

• 提供一个简单的命令行界面
• 处理用户输入并显示响应
• 包含基础错误处理
• 允许优雅退出

6.5 资源管理 #

• 资源的正确清理
• 连接问题的错误处理
• 优雅的关闭流程

7. 运行客户端 #

要在任何MCP服务器上运行您的客户端：

uv run client.py path/to/server.py    # python服务器
uv run client.py path/to/build/index.js  # node服务器

提示: 如果您正在继续从服务器快速入门教程中的天气示例，您的命令可能类似于这样： python client.py .../quickstart-resources/weather-server-python/weather.py

客户端将：

1. 连接到指定的服务器
2. 列出可用工具
3. 开启一个交互式聊天会话，您可以在此：
   • 输入查询
   • 查看工具执行情况
   • 获取Claude的回复

以下是连接到服务器快速入门中的天气服务后应有的示例效果：

8.⚙️ 运作原理 #

当您提交一个查询时：

1. 客户端从服务器获取可用工具列表
2. 您的查询会连同工具描述一并发送给Claude
3. Claude决定是否使用工具（以及使用哪些工具）
4. 客户端通过服务器执行所有请求的工具调用
5. 结果被发送回Claude
6. Claude提供了一个自然语言响应
7. 响应已显示给您

9.📚 最佳实践 #

9.1 错误处理 #

• 始终将工具调用包裹在try-catch代码块中
• 提供有意义的错误信息
• 优雅处理连接问题

9.2 资源管理 #

• 使用 AsyncExitStack 以便进行适当的清理
• 完成后关闭连接
• 处理服务器断开连接

9.3 安全 #

• 安全存储API密钥于 .env
• 验证服务器响应
• 谨慎处理工具权限

10.🔧 故障排除 #

10.1 服务器路径问题 #

• 请确认您的服务器脚本路径是否正确
• 如果相对路径无效，请使用绝对路径
• Windows用户请确保在路径中使用正斜杠(/)或转义反斜杠()
• 验证服务器文件是否具有正确的扩展名（Python为.py，Node.js为.js）

正确路径使用示例：

# 相对路径
uv run client.py ./server/weather.py

# 绝对路径
uv run client.py /Users/username/projects/mcp-server/weather.py

# Windows路径（两种格式都可以）
uv run client.py C:/projects/mcp-server/weather.py
uv run client.py C:\projects\mcp-server\weather.py

10.2 响应时间 #

• 首次响应可能需要长达30秒才能返回
• 这是正常现象，在以下情况下会发生：
  • 服务器初始化
  • Claude处理该查询
  • 正在执行工具
• 后续响应通常会更快
• 在此初始等待期间，请勿中断进程

10.3 常见错误信息 #

如果您看到：

• FileNotFoundError - 检查您的服务器路径
• Connection refused - 确保服务器正在运行且路径正确
• Tool execution failed - 验证工具的必需环境变量是否已设置
• Timeout error - 考虑在您的客户端配置中增加timeout