Skip to main content
本教程将教您如何构建一个连接到MCP服务器的LLM驱动聊天机器人客户端。建议您先完成服务器快速入门，它会指导您构建第一个服务器的基础知识。
您可以在这里找到本教程的完整代码。

​
系统要求

开始前，请确保您的系统满足以下要求：
  • 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 main.py

# 创建主文件
touch client.py

​
设置API密钥

您需要从Anthropic控制台获取Anthropic API密钥。创建.env文件来存储它：
# 创建.env文件
touch .env
将密钥添加到.env文件中：
ANTHROPIC_API_KEY=<您的密钥>
将.env添加到.gitignore：
echo ".env" >> .gitignore

​
创建客户端

​
基本客户端结构

首先，设置导入并创建基本客户端类：
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
    )

    # 处理响应和工具调用
    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)
            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
            )

            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 <服务器脚本路径>")
        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())

​
关键组件说明

​
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 服务器脚本路径.py # python服务器
uv run client.py 服务器脚本路径.js # node服务器
客户端将：
  1. 连接到指定的服务器
  2. 列出可用工具
  3. 开始交互式聊天会话，您可以：
    • 输入查询
    • 查看工具执行
    • 获取来自Claude的响应

​
工作原理

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

​
最佳实践

  1. 错误处理
    • 始终使用try-catch块包装工具调用
    • 提供有意义的错误消息
    • 优雅处理连接问题
  2. 资源管理
    • 使用AsyncExitStack进行适当清理
    • 使用完成后关闭连接
    • 处理服务器断开连接
  3. 安全性
    • 在.env中安全存储API密钥
    • 验证服务器响应
    • 谨慎处理工具权限

​
故障排除

​
服务器路径问题

  • 仔细检查服务器脚本路径是否正确
  • 如果相对路径不起作用，请使用绝对路径
  • 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集成的客户端列表

用LLMs构建MCP

了解如何使用像Claude这样的LLMs加速您的MCP开发

核心架构

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