为客户端开发者

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

在本教程中，您将学习如何构建一个连接到 MCP 服务器的 LLM 驱动的聊天机器人客户端。最好先阅读服务器快速入门，它会指导您构建第一个服务器的基础知识。

• Python

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

系统要求

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

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

设置您的环境

首先，使用 uv 创建一个新的 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

# 删除样板文件
rm hello.py

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

设置您的 API 密钥

您需要从 Anthropic 控制台 获取一个 Anthropic API 密钥。

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

# 创建 .env 文件
touch .env

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

ANTHROPIC_API_KEY=<your key here>

将 .env 添加到您的 .gitignore 中：

echo ".env" >> .gitignore

确保您的 ANTHROPIC_API_KEY 安全！

创建客户端

基本客户端结构

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

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

load_dotenv()  # 从 .env 加载环境变量

class MCPClient:
    def __init__(self):
        # 初始化会话和客户端对象
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
    # 方法将在这里添加

服务器连接管理

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

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
    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])

查询处理逻辑

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

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
    )

    # 处理响应并处理工具调用
    tool_results = []
    final_text = []

    for content in response.content:
        if content.type == 'text':
            final_text.append(content.text)
        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}]")

            # 使用工具结果继续对话
            if hasattr(content, 'text') and content.text:
                messages.append({
                  "role": "assistant",
                  "content": content.text
                })
            messages.append({
                "role": "user", 
                "content": result.content
            })

            # 获取 Claude 的下一个响应
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
            )

            final_text.append(response.content[0].text)

    return "\n".join(final_text)

交互式聊天界面

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

async def chat_loop(self):
    """运行交互式聊天循环"""
    print("\nMCP 客户端已启动！")
    print("输入您的查询或 'quit' 退出。")
    
    while True:
        try:
            query = input("\n查询: ").strip()
            
            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()

主入口点

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

async def main():
    if len(sys.argv) < 2:
        print("用法: 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. 客户端初始化

• MCPClient 类初始化会话管理和 API 客户端
• 使用 AsyncExitStack 进行适当的资源管理
• 配置 Anthropic 客户端以进行 Claude 交互

2. 服务器连接

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

3. 查询处理

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

4. 交互界面

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

5. 资源管理

• 适当清理资源
• 连接问题的错误处理
• 优雅的关闭程序

常见自定义点

1. 工具处理

   • 修改 process_query() 以处理特定工具类型
   • 为工具调用添加自定义错误处理
   • 实现工具特定的响应格式

2. 响应处理

   • 自定义工具结果的格式
   • 添加响应过滤或转换
   • 实现自定义日志记录

3. 用户界面

   • 添加 GUI 或 Web 界面
   • 实现丰富的控制台输出
   • 添加命令历史或自动完成

运行客户端

使用任何 MCP 服务器运行您的客户端：

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

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

客户端将：

1. 连接到指定的服务器
2. 列出可用工具
3. 启动交互式聊天会话，您可以：
   • 输入查询
   • 查看工具执行
   • 获取 Claude 的响应

以下是连接到服务器快速入门中的天气服务器时的示例：

工作原理

当您提交查询时：

1. 客户端从服务器获取可用工具列表
2. 您的查询连同工具描述一起发送到 Claude
3. Claude 决定使用哪些工具（如果有）
4. 客户端通过服务器执行任何请求的工具调用
5. 结果返回给 Claude
6. Claude 提供自然语言响应
7. 响应显示给您

最佳实践

1. 错误处理

   • 始终在 try-catch 块中包装工具调用
   • 提供有意义的错误消息
   • 优雅地处理连接问题

2. 资源管理

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

3. 安全性

   • 将 API 密钥安全地存储在 .env 中
   • 验证服务器响应
   • 对工具权限保持谨慎

故障排除

服务器路径问题

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

响应时间

• 第一个响应可能需要长达 30 秒的时间返回
• 这是正常的，发生在：
  • 服务器初始化
  • Claude 处理查询
  • 工具正在执行
• 后续响应通常更快
• 在此初始等待期间不要中断进程

常见错误消息

如果您看到：

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

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

系统要求

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

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

设置您的环境

首先，使用 uv 创建一个新的 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

# 删除样板文件
rm hello.py

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

设置您的 API 密钥

您需要从 Anthropic 控制台 获取一个 Anthropic API 密钥。

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

# 创建 .env 文件
touch .env

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

ANTHROPIC_API_KEY=<your key here>

将 .env 添加到您的 .gitignore 中：

echo ".env" >> .gitignore

确保您的 ANTHROPIC_API_KEY 安全！

创建客户端

基本客户端结构

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

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

load_dotenv()  # 从 .env 加载环境变量

class MCPClient:
    def __init__(self):
        # 初始化会话和客户端对象
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
    # 方法将在这里添加

服务器连接管理

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

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
    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])

查询处理逻辑

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

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
    )

    # 处理响应并处理工具调用
    tool_results = []
    final_text = []

    for content in response.content:
        if content.type == 'text':
            final_text.append(content.text)
        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}]")

            # 使用工具结果继续对话
            if hasattr(content, 'text') and content.text:
                messages.append({
                  "role": "assistant",
                  "content": content.text
                })
            messages.append({
                "role": "user", 
                "content": result.content
            })

            # 获取 Claude 的下一个响应
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
            )

            final_text.append(response.content[0].text)

    return "\n".join(final_text)

交互式聊天界面

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

async def chat_loop(self):
    """运行交互式聊天循环"""
    print("\nMCP 客户端已启动！")
    print("输入您的查询或 'quit' 退出。")
    
    while True:
        try:
            query = input("\n查询: ").strip()
            
            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()

主入口点

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

async def main():
    if len(sys.argv) < 2:
        print("用法: 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. 客户端初始化

• MCPClient 类初始化会话管理和 API 客户端
• 使用 AsyncExitStack 进行适当的资源管理
• 配置 Anthropic 客户端以进行 Claude 交互

2. 服务器连接

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

3. 查询处理

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

4. 交互界面

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

5. 资源管理

• 适当清理资源
• 连接问题的错误处理
• 优雅的关闭程序

常见自定义点

1. 工具处理

   • 修改 process_query() 以处理特定工具类型
   • 为工具调用添加自定义错误处理
   • 实现工具特定的响应格式

2. 响应处理

   • 自定义工具结果的格式
   • 添加响应过滤或转换
   • 实现自定义日志记录

3. 用户界面

   • 添加 GUI 或 Web 界面
   • 实现丰富的控制台输出
   • 添加命令历史或自动完成

运行客户端

使用任何 MCP 服务器运行您的客户端：

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

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

客户端将：

1. 连接到指定的服务器
2. 列出可用工具
3. 启动交互式聊天会话，您可以：
   • 输入查询
   • 查看工具执行
   • 获取 Claude 的响应

以下是连接到服务器快速入门中的天气服务器时的示例：

工作原理

当您提交查询时：

1. 客户端从服务器获取可用工具列表
2. 您的查询连同工具描述一起发送到 Claude
3. Claude 决定使用哪些工具（如果有）
4. 客户端通过服务器执行任何请求的工具调用
5. 结果返回给 Claude
6. Claude 提供自然语言响应
7. 响应显示给您

最佳实践

1. 错误处理

   • 始终在 try-catch 块中包装工具调用
   • 提供有意义的错误消息
   • 优雅地处理连接问题

2. 资源管理

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

3. 安全性

   • 将 API 密钥安全地存储在 .env 中
   • 验证服务器响应
   • 对工具权限保持谨慎

故障排除

服务器路径问题

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

响应时间

• 第一个响应可能需要长达 30 秒的时间返回
• 这是正常的，发生在：
  • 服务器初始化
  • Claude 处理查询
  • 工具正在执行
• 后续响应通常更快
• 在此初始等待期间不要中断进程

常见错误消息

如果您看到：

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

下一步

示例服务器

查看我们的官方 MCP 服务器和实现的画廊

客户端

查看支持 MCP 集成的客户端列表

使用 LLM 构建 MCP

了解如何使用 Claude 等 LLM 加速您的 MCP 开发

核心架构

了解 MCP 如何连接客户端、服务器和 LLM