Skip to main content
在本教程中，你将学习如何构建一个由 LLM 驱动、并能连接 MCP 服务器的聊天机器人客户端。 开始之前，建议先阅读构建 MCP 服务器教程，这有助于理解客户端和服务器如何通信。
这里可以找到本教程的完整代码。

​
系统要求

开始前，请确保系统满足以下要求：
  • Mac 或 Windows 电脑。
  • 已安装最新 Python 版本。
  • 已安装最新版本的 uv。

​
设置环境

首先使用 uv 创建新的 Python 项目：
# Create project directory
uv init mcp-client
cd mcp-client

# Create virtual environment
uv venv

# Activate virtual environment
source .venv/bin/activate

# Install required packages
uv add mcp anthropic python-dotenv

# Remove boilerplate files
rm main.py

# Create our main file
touch client.py

​
设置 API Key

你需要从 Anthropic Console 获取 Anthropic API key。创建 .env 文件来保存它：
echo "ANTHROPIC_API_KEY=your-api-key-goes-here" > .env
将 .env 添加到 .gitignore：
echo ".env" >> .gitignore
请确保妥善保护你的 ANTHROPIC_API_KEY！

​
创建客户端

​
基础客户端结构

首先设置 imports，并创建基础客户端类：
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 server 的方法：
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])

​
查询处理逻辑

现在添加处理查询和 tool calls 的核心功能：
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-sonnet-4-20250514",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )

    # Process response and handle tool calls
    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

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

            # Get next response from Claude
            response = self.anthropic.messages.create(
                model="claude-sonnet-4-20250514",
                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):
    """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. Client 初始化

  • MCPClient 类会初始化 session 管理和 API clients
  • 使用 AsyncExitStack 正确管理资源
  • 配置用于 Claude 交互的 Anthropic client

​
2. Server 连接

  • 同时支持 Python 和 Node.js servers
  • 校验 server script 类型
  • 设置正确的通信通道
  • 初始化 session 并列出可用 tools

​
3. 查询处理

  • 维护对话上下文
  • 处理 Claude 响应和 tool calls
  • 管理 Claude 与 tools 之间的消息流
  • 将结果组合成连贯响应

​
4. 交互界面

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

​
5. 资源管理

  • 正确清理资源
  • 处理连接问题
  • 执行优雅关闭流程

​
常见自定义点

  1. Tool 处理
    • 修改 process_query() 以处理特定 tool 类型
    • 为 tool calls 添加自定义错误处理
    • 实现特定 tool 的响应格式化
  2. 响应处理
    • 自定义 tool results 的格式化方式
    • 添加响应过滤或转换
    • 实现自定义日志
  3. 用户界面
    • 添加 GUI 或 Web 界面
    • 实现富控制台输出
    • 添加命令历史或自动补全

​
运行 Client

要让 client 连接任意 MCP server：
uv run client.py path/to/server.py # python server
uv run client.py path/to/build/index.js # node server
如果你继续使用 server quickstart 中的天气教程，命令可能类似这样：python client.py .../quickstart-resources/weather-server-python/weather.py
client 将会：
  1. 连接到指定 server
  2. 列出可用 tools
  3. 启动交互式聊天 session，你可以：
    • 输入查询
    • 查看 tool 执行
    • 获取 Claude 的响应
如果连接到 server quickstart 中的天气 server，效果大致如下：

​
工作原理

提交查询时：
  1. client 从 server 获取可用 tools 列表
  2. 你的查询会连同 tool 描述一起发送给 Claude
  3. Claude 决定使用哪些 tools（如果需要）
  4. client 通过 server 执行任何被请求的 tool calls
  5. 结果发回 Claude
  6. Claude 提供自然语言响应
  7. 响应显示给你

​
最佳实践

  1. 错误处理
    • 始终用 try-catch blocks 包裹 tool calls
    • 提供有意义的错误消息
    • 优雅处理连接问题
  2. 资源管理
    • 使用 AsyncExitStack 正确清理
    • 完成后关闭连接
    • 处理 server 断开连接
  3. 安全
    • 将 API keys 安全存储在 .env 中
    • 校验 server 响应
    • 谨慎处理 tool 权限
  4. Tool 名称
    • Tool 名称可以根据这里指定的格式进行校验
    • 如果 tool 名称符合指定格式，MCP client 不应使其校验失败

​
排障

​
Server 路径问题

  • 仔细检查 server script 路径是否正确
  • 如果相对路径不工作，请使用绝对路径
  • Windows 用户请确保路径中使用正斜杠 (/) 或转义反斜杠 (\)
  • 确认 server 文件扩展名正确（Python 用 .py，Node.js 用 .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 秒才返回
  • 这是正常现象，期间会发生：
    • server 初始化
    • Claude 处理查询
    • tools 正在执行
  • 后续响应通常更快
  • 在初始等待期间不要中断进程

​
常见错误消息

如果看到：
  • FileNotFoundError：检查 server 路径
  • Connection refused：确保 server 正在运行且路径正确
  • Tool execution failed：确认 tool 所需环境变量已设置
  • Timeout error：考虑在 client 配置中增加 timeout

​
后续步骤

示例服务器

查看官方 MCP servers 和 implementations 展示列表