客户端开发

开始构建可与所有 MCP 服务器集成的您自己的客户端。

在本教程中，您将学习如何构建连接到 MCP 服务器的 LLM 支持的聊天机器人客户端。完成 Server 快速入门会有所帮助，该快速入门将指导您完成构建第一个 Server 的基础知识。

您可以在此处找到本教程的完整代码。

一、系统要求

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

Mac 或 Windows 计算机
已安装最新的 Python 版本
最新版本的已安装uv

二、设置环境

首先，使用：uv

bash

# 创建项目目录
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

# 删除默认的示例文件
rm hello.py

# 创建主程序文件
touch client.py

三、设置 API 密钥

您需要来自 Anthropic 控制台的 Anthropic API 密钥。

创建一个文件来存储它：.env

bash

# 创建 .env 文件
touch .env

将密钥添加到文件中：.env

bash

ANTHROPIC_API_KEY=<your key here>

添加到您的：.env``.gitignore

bash

echo ".env" >> .gitignore

确保你的安全！ANTHROPIC_API_KEY

四、创建客户端

1、基本客户端结构

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

python

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

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

class MCPClient:
    def __init__(self):
        """初始化 MCP 客户端，创建会话和客户端对象。"""
        self.session: Optional[ClientSession] = None  # MCP 客户端会话对象
        self.exit_stack = AsyncExitStack()  # 管理异步资源的退出栈
        self.anthropic = Anthropic()  # 初始化 Anthropic API 客户端

    # 其他方法将在此处添加

2、服务器连接管理

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

python

async def connect_to_server(self, server_script_path: str):
    """连接到 MCP 服务器

    参数:
        server_script_path: 服务器脚本的路径 (.py 或 .js)
    """
    is_python = server_script_path.endswith('.py')
    is_js = server_script_path.endswith('.js')
    if not (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

    # 创建 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("\n已连接到服务器，可用工具:", [tool.name for tool in tools])

3、查询处理逻辑

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

python

async def process_query(self, query: str) -> str:
    """使用 Claude 和可用工具处理查询"""

    # 初始化用户消息
    messages = [
        {
            "role": "user",
            "content": query
        }
    ]

    # 获取服务器提供的工具信息
    response = await self.session.list_tools()
    available_tools = [{
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    } for tool in response.tools]

    # 调用 Claude API 进行初步处理
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )

    # 处理 Claude 响应并执行工具调用
    tool_results = []  # 存储工具调用结果
    final_text = []  # 存储最终返回的文本

    assistant_message_content = []  # 存储助理的消息内容
    for content in response.content:
        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

            # 执行工具调用
            result = await self.session.call_tool(tool_name, tool_args)
            tool_results.append({"call": tool_name, "result": result})
            final_text.append(f"[调用工具 {tool_name}，参数 {tool_args}]")

            # 记录助理的工具调用请求
            assistant_message_content.append(content)
            messages.append({
                "role": "assistant",
                "content": assistant_message_content
            })

            # 记录工具的执行结果
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": content.id,
                        "content": result.content
                    }
                ]
            })

            # 获取 Claude 的后续响应
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )

            # 追加 Claude 返回的文本信息
            final_text.append(response.content[0].text)

    return "\n".join(final_text)

4、交互式聊天界面

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

python

async def chat_loop(self):
    """运行交互式聊天循环"""

    print("\nMCP 客户端已启动！")
    print("请输入您的查询，或输入 'quit' 退出。")

    while True:
        try:
            # 获取用户输入
            query = input("\nQuery: ").strip()

            # 如果用户输入 'quit'，则退出循环
            if query.lower() == 'quit':
                break

            # 处理用户查询并获取响应
            response = await self.process_query(query)

            # 输出响应内容
            print("\n" + response)

        except Exception as e:
            # 处理并打印异常
            print(f"\n错误: {str(e)}")


async def cleanup(self):
    """清理资源"""
    await self.exit_stack.aclose()

5、主要入口

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

python

async def main():
    if len(sys.argv) < 2:
        print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    import sys
    asyncio.run(main())

您可以在此处找到完整的文件。client.py

五、关键组件解释

1. 客户端初始化

该类使用会话管理和 API 客户端进行初始化MCPClient
正确资源管理的用途AsyncExitStack
为 Claude 交互配置 Anthropic 客户端

2. 服务器连接

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

3. 查询处理

维护对话上下文
处理 Claude 的响应和工具调用
管理 Claude 和 tools 之间的消息流
将结果合并为连贯的响应

4. 交互式界面

提供简单的命令行界面
处理用户输入并显示响应
包括基本错误处理
允许正常退出

5. 资源管理

正确清理资源
连接问题的错误处理
正常关闭过程

六、常见自定义点

工具处理
- 修改以处理特定工具类型process_query()
- 为工具调用添加了自定义错误处理
- 实施特定于工具的响应格式
响应处理
- 自定义工具结果的格式
- 添加响应筛选或转换
- 实施自定义日志记录
用户界面
- 添加 GUI 或 Web 界面
- 实现丰富的控制台输出
- 添加命令历史记录或自动完成

七、运行客户端

要运行客户端，传入任何 MCP 服务器：

bash

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

如果您继续服务器快速入门中的天气教程，您的命令可能如下所示：python client.py .../weather/src/weather/server.py

客户将：

连接到指定的服务器
列出可用工具
开始交互式聊天会话，您可以在其中：
- 输入查询
- 查看工具执行
- 获取 Claude 的回复

下面是一个示例，说明从服务器快速入门连接到 weather 服务器时应是什么样子：

八、如何运作

当您提交查询时：

客户端从服务器获取可用工具的列表
您的查询将与工具描述一起发送给 Claude
Claude 决定使用哪些工具（如果有）
客户端通过服务器执行任何请求的工具调用
结果将发送回 Claude
Claude 提供自然语言响应
此时将向您显示响应

其他

1、最佳实践

错误处理
- 始终将工具调用包装在 try-catch 块中
- 提供有意义的错误消息
- 妥善处理连接问题
资源管理
- 用于正确清理AsyncExitStack
- 完成后关闭连接
- 处理服务器断开连接
安全
- 将 API 密钥安全地存储在.env
- 验证服务器响应
- 谨慎使用工具权限

2、故障排除

服务器路径问题

仔细检查服务器脚本的路径是否正确
如果相对路径不起作用，请使用绝对路径
对于 Windows 用户，请确保在路径中使用正斜杠（/）或转义的反斜杠（\）
验证服务器文件是否具有正确的扩展名（Python 的扩展名.py 或 Node.js 的扩展名.js）

正确路径使用示例：

bash

# 相对路径
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

响应时序

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

常见错误消息

如果您看到：

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

客户端开发 ​

一、系统要求 ​

二、设置环境 ​

三、设置 API 密钥 ​

四、创建客户端 ​

1、基本客户端结构 ​

2、服务器连接管理 ​

3、查询处理逻辑 ​

4、交互式聊天界面 ​

5、主要入口 ​

五、关键组件解释 ​

1. 客户端初始化 ​

2. 服务器连接 ​

3. 查询处理 ​

4. 交互式界面 ​

5. 资源管理 ​

六、常见自定义点 ​

七、运行客户端 ​

八、如何运作 ​

其他 ​

1、最佳实践 ​

2、故障 排除 ​

服务器路径问题 ​

响应时序 ​

常见错误消息 ​

客户端开发

一、系统要求

二、设置环境

三、设置 API 密钥

四、创建客户端

1、基本客户端结构

2、服务器连接管理

3、查询处理逻辑

4、交互式聊天界面

5、主要入口

五、关键组件解释

1. 客户端初始化

2. 服务器连接

3. 查询处理

4. 交互式界面

5. 资源管理

六、常见自定义点

七、运行客户端

八、如何运作

其他

1、最佳实践

2、故障排除

服务器路径问题

响应时序

常见错误消息