MCP客户端Client开发流程

1. uv工具入门使用指南

1.1 uv入门介绍

MCP开发要求借助uv进行虚拟环境创建和依赖管理。 uv 是一个Python 依赖管理工具，类似于pip 和 conda ，但它更快、更高效，并且可以更好地管理 Python 虚拟环境和依赖项。它的核心目标是 替代 pip 、 venv 和 pip-tools ，提供更好的性能和更低的管理开销。

uv 的特点：

1. 速度更快：相比 pip ， uv 采用 Rust 编写，性能更优。
2. 支持 PEP 582：无需 virtualenv ，可以直接使用 __pypackages__ 进行管理。
3. 兼容 pip ：支持 requirements.txt 和 pyproject.toml 依赖管理。
4. 替代 venv ：提供 uv venv 进行虚拟环境管理，比 venv 更轻量。
5. 跨平台：支持 Windows、macOS 和 Linux。

1.2 uv安装流程

方法 1：使用 pip 安装（适用于已安装 pip 的系统）

pip install uv

方法 2：使用 curl 直接安装

如果你的系统没有 pip ，可以直接运行：

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

这会自动下载 uv 并安装到 /usr/local/bin 。

1.3 uv的基本用法介绍

安装 uv 后，你可以像 pip 一样使用它，但它的语法更简洁，速度也更快。注意，以下为使用语法

示例，不用实际运行。

• 安装 Python 依赖

uv pip install requests

与 pip install requests 类似，但更快。

• 创建虚拟环境

uv venv myenv

等效于 python -m venv myenv ，但更高效。

• 激活虚拟环境

source myenv/bin/activate     # Linux/macOS

myenv\Scripts\activate          # Windows

• 安装 requirements.txt

uv pip install -r requirements.txt

• 直接运行 Python 项目

如果项目中包含 pyproject.toml ，你可以直接运行：

uv run python script.py

这等效于：

pip install -r requirements.txt

python script.py

但 uv 速度更快，管理更高效。

为什么 MCP 更推荐使用 uv 进行环境管理？

MCP 依赖的 Python 环境可能包含多个模块， uv 通过 pyproject.toml 提供更高效的管理方 式，并且可以避免 pip 的一些依赖冲突问题。此外， uv 的包管理速度远超 pip ，这对于 MCP 这样频繁管理依赖的项目来说是一个很大的优势。

2.MCP极简客户端搭建流程

2.1 创建 MCP 客户端项目

# 创建项目目录

uv init mcp-client

cd mcp-client

2.2 创建MCP客户端虚拟环境

# 创建虚拟环境

uv venv

# 激活虚拟环境

source .venv/bin/activate

这里需要注意的是，相比 pip ， uv 会自动识别当前项目主目录并创建虚拟环境。

然后即可通过 add 方法在虚拟环境中安装相关的库。

# 安装 MCP SDK

uv add mcp

2.3 编写基础 MCP 客户端

然后在当前项目主目录中创建 client.py

并写入以下代码

import asyncio # 让代码支持异步操作
from mcp import ClientSession # MCP 客户端会话管理
from contextlib import AsyncExitStack # 资源管理（确保客户端关闭时释放资源）class MCPClient:def __init__(self):"""初始化 MCP 客户端"""self.session = None # 先不连接 MCP 服务器self.exit_stack = AsyncExitStack() # 创建资源管理器async def connect_to_mock_server(self):"""模拟 MCP 服务器的连接（暂不连接真实服务器）"""print("✅ MCP 客户端已初始化，但未连接到服务器")async def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP 客户端已启动！输入 'quit' 退出")while True:  # 无限循环，直到用户输入 'quit'try: query = input("\nQuery: ").strip() # 让用户输入问题if query.lower() == 'quit': # 如果用户输入 quit，退出循环breakprint(f"\n🤖 [Mock Response] 你说的是：{query}") # 返回模拟回复except Exception as e: # 发生错误时捕获异常print(f"\n⚠️ 发生错误: {str(e)}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose() # 关闭资源管理器async def main():client = MCPClient() # 创建 MCP 客户端try: await client.connect_to_mock_server() # 连接（模拟）服务器await client.chat_loop() # 进入聊天循环finally:await client.cleanup() # 确保退出时清理资源if __name__ == "__main__": #确保代码只能在 Python 直接运行时执行（而不是作为库导入时）asyncio.run(main())

这段代码能够初始化 MCP 客户端（但不连接服务器），并提供一个 交互式 CLI ，可以输入查询（但只返回模拟回复），通过输入 quit 退出程序 。需要注意的是，此时客户端没有关联任何大模型，因此只会 重复用户的输入。

MCP 中一个基础的客户端代码结构 总结如下 ：

代码部分	作用
MCPClient.__init__()	初始化 MCP 客户端
connect_to_mock_server()	模拟 MCP 服务器连接
chat_loop()	提供交互式聊天界面
cleanup()	释放资源
main()	启动客户端
asyncio.run(main())	运行程序

2.4 运行 MCP 客户端

然后尝试运行这个极简的 MCP 客户端：

uv run client.py

3. MCP客户端接入DeepSeek在线模型流程

3.1 新增依赖

为了支持调用 DeepSeek 模型，以及在环境变量中读取 API-KEY 等信息，需要先安装如下依赖：

uv add mcp openai python-dotenv

3.2 创建.env文件

写入如下内容

BASE_URL = https://api.deepseek.com

MODEL = deepseek-chat

OPENAI_API_KEY = "DeepSeek API-Key"

3.3 修改client.py代码

import asyncio
import os
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()class MCPClient:def  __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Keyself.base_url = os.getenv("BASE_URL")  # 读取 BASE YRLself.model = os.getenv("MODEL")  # 读取 modelif not self.openai_api_key:raise ValueError("❌ 未找到 OpenAI API Key，请在 .env 文件中设置OPENAI_API_KEY")self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)async def process_query(self, query: str) -> str:"""调用 DeepSeek API 处理用户查询"""messages = [{"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},{"role": "user", "content": query}]try:# 调用 OpenAI APIresponse = await asyncio.get_event_loop().run_in_executor(None,lambda: self.client.chat.completions.create(model=self.model,messages=messages))return response.choices[0].message.contentexcept Exception as e:return f"⚠️ 调用 DeepSeek API 时出错: {str(e)}"async def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP 客户端已启动！输入 'quit' 退出")while True:try:query = input("\nQuery: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)  # 发送用户输入到 OpenAIAPIprint(f"\n🤖 DeepSeek: {response}")except Exception as e:print(f"\n⚠️ 发生错误: {str(e)}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():client = MCPClient()try:await client.chat_loop()finally:await client.cleanup()
if __name__ == "__main__":asyncio.run(main())