1. 你将构建什么？ #

一个能连接 MCP 服务器的聊天客户端：你输入问题，它通过 DeepSeek 理解你的意图，自动调用 MCP 服务器提供的工具（如查天气、读文件等），并把结果用自然语言回复给你。

能力	说明
连接 MCP 服务器	支持 Python 或 Node.js 编写的服务器
调用工具	自动获取工具列表并执行 DeepSeek 选中的工具
对话式交互	命令行输入问题，获取自然语言回复

完整示例代码

2. 前置要求 #

要求	说明
Mac 或 Windows	任意常见操作系统
Python 3.10+	需已安装
uv	Python 包管理工具，本教程会用到
Anthropic API 密钥	用于调用 DeepSeek，在此获取

零基础提示：若不了解「异步」「会话」等概念，可先按步骤操作，遇到不懂的再查阅；代码中会尽量用注释说明每步在做什么。

3. 整体流程一览 #

你输入问题 → 客户端发给 DeepSeek → DeepSeek 决定是否调用工具
                                    ↓
            客户端执行工具 ← 向 MCP 服务器发起调用
                                    ↓
            结果返回 DeepSeek → DeepSeek 生成自然语言回复 → 显示给你

先建立整体印象，后面会按步骤实现。

4. 环境准备 #

4.1 创建项目 #

在终端中执行（macOS/Linux）：

# 创建项目目录
uv init mcp-client
cd mcp-client

# 创建虚拟环境
uv venv

# 激活虚拟环境
source .venv/bin/activate

# 安装依赖：mcp（MCP 协议）、anthropic（DeepSeek API）、python-dotenv（读取 .env）
uv add mcp anthropic python-dotenv

# 删除默认生成的 main.py
rm main.py

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

Windows（PowerShell）：

# 创建项目目录
uv init mcp-client
cd mcp-client

# 创建虚拟环境
uv venv

# 激活虚拟环境
.venv\Scripts\activate

# 安装依赖
uv add mcp anthropic python-dotenv

# 删除默认生成的 main.py
del main.py

# 创建主程序文件
new-item client.py

4.2 配置 API 密钥 #

1. 在 Anthropic 控制台 创建 API 密钥
2. 在项目根目录创建 .env 文件，写入：

ANTHROPIC_API_KEY=你的密钥

3. 将 .env 加入 .gitignore，避免误提交：

echo ".env" >> .gitignore

安全提示：请勿将 API 密钥提交到 Git 或分享给他人。

5. 分步实现客户端 #

下面按功能模块逐步编写代码，每步都有简要说明。

5.1 基本结构：导入与初始化 #

打开 client.py，先写入导入和 MCPClient 类的基本结构：

# 异步相关
import asyncio
import sys
from typing import Optional
from contextlib import AsyncExitStack

# MCP 客户端与 stdio 连接
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# DeepSeek API 与环境变量
from anthropic import Anthropic
from dotenv import load_dotenv

# 从 .env 加载环境变量（含 ANTHROPIC_API_KEY）
load_dotenv()


class MCPClient:
    """MCP 客户端：连接服务器、调用工具、与 DeepSeek 对话。"""

    def __init__(self):
        # 会话对象，连接建立后才有值
        self.session: Optional[ClientSession] = None
        # 用于统一管理异步资源的退出（连接、会话等）
        self.exit_stack = AsyncExitStack()
        # Anthropic 客户端，用于调用 DeepSeek API
        self.anthropic = Anthropic()

5.2 连接 MCP 服务器 #

在 MCPClient 类中添加连接方法：

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

    Args:
        server_script_path: 服务器脚本路径（.py 或 .js）
    """
    # 根据扩展名判断用 python 还是 node 启动
    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 连接（标准输入/输出与服务器通信）
    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])

5.3 处理用户查询（核心逻辑） #

实现「用户问题 → DeepSeek → 工具调用 → 最终回复」的流程：

async def process_query(self, query: str) -> str:
    """处理用户查询：交给 DeepSeek，必要时调用工具，返回最终回复。"""
    messages = [
        {"role": "user", "content": query}
    ]

    # 从 MCP 服务器获取工具列表
    response = await self.session.list_tools()
    available_tools = [
        {
            "name": tool.name,
            "description": tool.description,
            "input_schema": tool.inputSchema
        }
        for tool in response.tools
    ]

    # 第一次调用 DeepSeek，传入用户问题和工具列表
    response = self.anthropic.messages.create(
        model="DeepSeek-sonnet-4-20250514",
        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':
            # DeepSeek 请求调用工具
            tool_name = content.name
            tool_args = content.input

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

            # 把 assistant 的 tool_use 加入消息历史
            assistant_message_content.append(content)
            messages.append({"role": "assistant", "content": assistant_message_content})
            # 把工具结果作为 user 消息反馈给 DeepSeek
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": content.id,
                        "content": result.content
                    }
                ]
            })

            # 再次调用 DeepSeek，让它根据工具结果生成回复
            response = self.anthropic.messages.create(
                model="DeepSeek-sonnet-4-20250514",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )

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

    return "\n".join(final_text)

5.4 交互式聊天与资源清理 #

添加聊天循环和退出时的清理逻辑：

async def chat_loop(self):
    """运行交互式聊天：循环接收输入，直到输入 quit。"""
    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()

5.5 主入口 #

在文件末尾添加主函数和入口：

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__":
    asyncio.run(main())

完整 client.py 可参考 GitHub 示例。

6. 运行客户端 #

6.1 启动命令 #

需要先有一个 MCP 服务器。若已按 构建 MCP 服务器 完成天气服务器，可这样运行：

# 连接 Python 服务器
uv run client.py path/to/weather.py

# 连接 Node.js 服务器
uv run client.py path/to/build/index.js

6.2 运行效果 #

客户端会：

1. 连接到指定服务器
2. 打印可用工具列表
3. 进入交互式聊天，你可以：
   • 输入问题（如「加州有什么天气警报？」）
   • 看到工具调用过程
   • 收到 DeepSeek 的自然语言回复

示例截图：

7. 工作原理简述 #

一次查询的大致流程：

1. 客户端从 MCP 服务器获取工具列表
2. 将你的问题 + 工具描述一起发给 DeepSeek
3. DeepSeek 决定是否调用工具、调用哪些、传什么参数
4. 客户端通过 MCP 协议向服务器执行工具调用
5. 工具结果返回给 DeepSeek
6. DeepSeek 根据结果生成自然语言回复
7. 客户端把回复展示给你

8. 关键概念速查 #

概念	说明
ClientSession	MCP 客户端会话，负责与服务器通信、调用工具
StdioServerParameters	描述如何用 stdio 启动并连接服务器进程
AsyncExitStack	统一管理多个异步上下文，退出时自动清理
tools / tool_use	DeepSeek 可「请求调用工具」，客户端负责真正执行

9. 常见自定义方向 #

• 工具处理：在 process_query 中为特定工具加错误处理或格式化
• 响应展示：自定义工具结果的展示方式、日志输出
• 界面：用 Web 或 GUI 替代命令行，或增加历史、补全等

10. 故障排除 #

10.1 服务器路径问题 #

• 确认服务器脚本路径正确（可用绝对路径测试）
• Windows 路径可用 / 或 \\
• 扩展名需为 .py（Python）或 .js（Node.js）

# 相对路径
uv run client.py ./server/weather.py

# 绝对路径（示例）
uv run client.py C:/projects/mcp-server/weather.py

10.2 首次响应较慢 #

• 首次请求可能需要 10–30 秒（服务器启动、DeepSeek 推理、工具执行）
• 后续请求通常更快
• 等待时不要频繁中断进程

10.3 常见错误 #

错误	可能原因
FileNotFoundError	服务器脚本路径错误
Connection refused	服务器未正确启动或路径不对
Tool execution failed	工具依赖的环境变量未设置
Timeout error	可尝试增加客户端超时配置

11. 安全与最佳实践 #

1. API 密钥：放在 .env，不要提交到 Git
2. 错误处理：工具调用建议用 try-except 包裹，给出清晰错误提示
3. 资源清理：使用 AsyncExitStack，确保连接正确关闭
4. 工具权限：谨慎决定客户端可调用哪些工具、对哪些服务器开放