1. uv工具入门使用指南
1.1 uv入门介绍
MCP开发要求借助uv进行虚拟环境创建和依赖管理。 uv 是一个Python 依赖管理工具,类似于pip 和 conda ,但它更快、更高效,并且可以更好地管理 Python 虚拟环境和依赖项。它的核心目标是 替代 pip 、 venv 和 pip-tools ,提供更好的性能和更低的管理开销。
uv 的特点:
- 速度更快:相比 pip , uv 采用 Rust 编写,性能更优。
- 支持 PEP 582:无需 virtualenv ,可以直接使用 __pypackages__ 进行管理。
- 兼容 pip :支持 requirements.txt 和 pyproject.toml 依赖管理。
- 替代 venv :提供 uv venv 进行虚拟环境管理,比 venv 更轻量。
- 跨平台:支持 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/macOSmyenv\Scripts\activate # Windows
- 安装 requirements.txt
uv pip install -r requirements.txt
- 直接运行 Python 项目
如果项目中包含
pyproject.toml
,你可以直接运行:
uv run python script.py
这等效于:
pip install -r requirements.txtpython script.py
但 uv 速度更快,管理更高效。
为什么
MCP
更推荐使用
uv
进行环境管理?
MCP
依赖的
Python
环境可能包含多个模块,
uv
通过
pyproject.toml
提供更高效的管理方 式,并且可以避免 pip
的一些依赖冲突问题。此外,
uv
的包管理速度远超
pip
,这对于
MCP 这样频繁管理依赖的项目来说是一个很大的优势。
2.MCP极简客户端搭建流程
2.1 创建 MCP 客户端项目
# 创建项目目录uv init mcp-clientcd mcp-client
2.2 创建MCP客户端虚拟环境
# 创建虚拟环境uv venv# 激活虚拟环境source .venv/bin/activate
这里需要注意的是,相比
pip
,
uv
会自动识别当前项目主目录并创建虚拟环境。
然后即可通过
add
方法在虚拟环境中安装相关的库。
# 安装 MCP SDKuv 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,退出循环
break
print(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.comMODEL = deepseek-chatOPENAI_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 Key
self.base_url = os.getenv("BASE_URL") # 读取 BASE YRL
self.model = os.getenv("MODEL") # 读取 model
if 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 API
response = 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.content
except 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':
break
response = await self.process_query(query) # 发送用户输入到 OpenAIAPI
print(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())