MCP 客户端Python开发指南

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

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

系统要求

开始之前，请确保您的系统符合以下要求：

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

uv是一个打包和项目管理的工具：https://github.com/astral-sh/uv
mac和linux安装方法：

curl -LsSf https://astral.sh/uv/install.sh | sh

windows安装方法：

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

设置您的环境

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

# Create project directory
uv init mcp-client
cd mcp-client

# Create virtual environment
uv venv

# Activate virtual environment
# On Windows:
.venv\Scripts\activate
# On Unix or MacOS:
source .venv/bin/activate

# Install required packages
uv add mcp anthropic python-dotenv

# Remove boilerplate files
rm hello.py

# Create our main file
touch client.py

设置您的 API 密钥

您需要从 Anthropic 控制台获取 Anthropic API 密钥。
创建一个 .env 文件来存储密钥：

# Create .env file
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()  # load environment variables from .env

class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
    # methods will go here

服务器连接管理

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

async def connect_to_server(self, server_script_path: str):
    """Connect to an MCP server
    
    Args:
        server_script_path: Path to the server script (.py or .js)
    """
    is_python = server_script_path.endswith('.py')
    is_js = server_script_path.endswith('.js')
    if not (is_python or is_js):
        raise ValueError("Server script must be a .py or .js file")
        
    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()
    
    # List available tools
    response = await self.session.list_tools()
    tools = response.tools
    print("\nConnected to server with tools:", [tool.name for tool in tools])

查询处理逻辑

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

async def process_query(self, query: str) -> str:
    """Process a query using Claude and available tools"""
    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]

    # Initial Claude API call
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )

    # Process response and handle tool calls
    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
            
            # Execute tool call
            result = await self.session.call_tool(tool_name, tool_args)
            tool_results.append({"call": tool_name, "result": result})
            final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

            # Continue conversation with tool results
            if hasattr(content, 'text') and content.text:
                messages.append({
                  "role": "assistant",
                  "content": content.text
                })
            messages.append({
                "role": "user", 
                "content": result.content
            })

            # Get next response from 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):
    """Run an interactive chat loop"""
    print("\nMCP Client Started!")
    print("Type your queries or 'quit' to exit.")
    
    while True:
        try:
            query = input("\nQuery: ").strip()
            
            if query.lower() == 'quit':
                break
                
            response = await self.process_query(query)
            print("\n" + response)
                
        except Exception as e:
            print(f"\nError: {str(e)}")

async def cleanup(self):
    """Clean up resources"""
    await self.exit_stack.aclose()

主入口

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

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. 客户端初始化

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

2. 服务器连接

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

3. 查询处理

• 保持对话上下文
• 处理克劳德的响应和工具调用
• 管理 Claude 和工具之间的消息流
• 将结果合并为连贯的回应

4. 交互界面

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

5. 资源管理

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

常见的定制点

1.工具操作

• 修改 process_query() 以处理特定的工具类型
• 为工具调用添加自定义错误处理
• 实现特定工具的响应格式化
  2.响应处理
• 自定义工具结果的格式化方式
• 添加响应过滤或转换
• 实现自定义日志记录
  3.用户界面
• 添加一个图形用户界面或 Web 界面
• 实现丰富的控制台输出
• 添加命令历史记录或自动完成

运行客户端

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

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

客户将：

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）

正确路径使用示例:

# Relative path
uv run client.py ./server/weather.py

# Absolute path
uv run client.py /Users/username/projects/mcp-server/weather.py

# Windows path (either format works)
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 ：考虑增加客户端配置中的超时时间