二、MCP客户端Client开发流程

环境要求:
当前MCP要求python在3.10以上版本,且MCP开发要求借助uv进行虚拟环境创建和依赖管理。

1. uv工具入门使用指南

1.1 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安装流程

使用pip安装:

pip install uv
C:\Users\xxxx>pip install uv
Collecting uv
  Downloading uv-0.7.2-py3-none-win_amd64.whl.metadata (11 kB)
  Downloading uv-0.7.2-py3-none-win_amd64.whl (18.3 MB)
   ---------------------------------------- 18.3/18.3 MB 1.7 MB/s eta 0:00:00
Installing collected packages: uv
Successfully installed uv-0.7.2

[notice] A new release of pip is available: 24.0 -> 25.1.1
[notice] To update, run: python.exe -m pip install --upgrade pip
1.3 uv的基本用法介绍
  • 安装 Python 依赖
uv pip install requests
  • 创建虚拟环境
uv venv myenv
  • 激活虚拟环境
source myenv/bin/activate  # Linux/macOS
myenv\Scripts\activate   # Windows
  • 直接运行 Python 项目
    如果项目中包含 pyproject.toml ,你可以直接运行:
uv run script.py

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

2.1 创建 MCP 客户端项目
# 创建项目目录
uv init weather
cd weather
D:\learning\mcp>uv init weather
Initialized project `weather` at `D:\learning\mcp\weather`

D:\learning\mcp>dir
 驱动器 D 中的卷是 Data
 卷的序列号是 6668-32CF

 D:\learning\mcp 的目录

2025/05/05  11:55    <DIR>          .
2025/05/05  10:36    <DIR>          ..
2025/05/05  11:55    <DIR>          weather
               0 个文件              0 字节
               4 个目录 287,324,241,920 可用字节

D:\learning\mcp>
D:\learning\mcp>cd weather

D:\learning\mcp\weather>dir
 驱动器 D 中的卷是 Data
 卷的序列号是 6668-32CF

 D:\learning\mcp\weather 的目录

2025/05/05  11:55    <DIR>          .
2025/05/05  11:55    <DIR>          ..
2025/05/05  11:55                 5 .python-version
2025/05/05  11:55                85 main.py
2025/05/05  11:55               153 pyproject.toml
2025/05/05  11:55                 0 README.md
               4 个文件            243 字节
               2 个目录 287,324,241,920 可用字节
2.2 创建MCP客户端虚拟环境
# 创建虚拟环境
uv venv
# 激活虚拟环境
source .venv/bin/activate   # linux
.venv\Scripts\activate         # windows
D:\learning\mcp\weather>uv venv
Using CPython 3.12.3 interpreter at: C:\Users\pq449\AppData\Local\Programs\Python\Python312\python.exe
Creating virtual environment at: .venv
Activate with: .venv\Scripts\activate

D:\learning\mcp\weather>

激活后进入mcp模式:

(weather) D:\learning\mcp\weather>
# 安装 MCP SDK
uv add mcp
Prepared 22 packages in 875ms
░░░░░░░░░░░░░░░░░░░░ [0/22] Installing wheels...                                                                        warning: Failed to hardlink files; falling back to full copy. This may lead to degraded performance.
         If the cache and target directories are on different filesystems, hardlinking may not be supported.
         If this is intentional, set `export UV_LINK_MODE=copy` or use `--link-mode=copy` to suppress this warning.
Installed 22 packages in 109ms
 + annotated-types==0.7.0
 + anyio==4.9.0
 + certifi==2025.4.26
 + click==8.1.8
 + colorama==0.4.6
 + h11==0.16.0
 + httpcore==1.0.9
 + httpx==0.28.1
 + httpx-sse==0.4.0
 + idna==3.10
 + mcp==1.7.1
 + pydantic==2.11.4
 + pydantic-core==2.33.2
 + pydantic-settings==2.9.1
 + python-dotenv==1.1.0
 + python-multipart==0.0.20
 + sniffio==1.3.1
 + sse-starlette==2.3.4
 + starlette==0.46.2
 + typing-extensions==4.13.2
 + typing-inspection==0.4.0
 + uvicorn==0.34.2

(weather) D:\learning\mcp\weather>
2.3 编写基础 MCP 客户端

使用vscode打开上面已经创建的项目,在项目内新建一个python文件,命名为client1.py,并输入下面内容


image.png
import asyncio
from contextlib import AsyncExitStack

class MCPClient:
    def __init__(self):
        """初始化 MCP 客户端"""
        self.session = None
        self.exit_stack = AsyncExitStack()
    
    async def connect_to_mock_server(self):
        """模拟 MCP 服务器的连接(暂不连接真实服务器)"""
        print("✅ MCP 客户端已初始化,但未连接到服务器")
    
    async def chat_loop(self):
        """运行交互式聊天循环"""
        print("\nMCP 客户端已启动!输入 'quit' 退出")
        while True:
            try:
                query = input("\nQuery: ").strip()
                if query.lower() == '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()

    try:
        await client.connect_to_mock_server()
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

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

运行效果如下:
在vscode的Terminal下面创建并激活虚拟环境:

PS D:\learning\mcp\weather> uv venv
Using CPython 3.12.3 interpreter at: C:\Users\pq449\AppData\Local\Programs\Python\Python312\python.exe
Creating virtual environment at: .venv
Activate with: .venv\Scripts\activate
PS D:\learning\mcp\weather> .\.venv\Scripts\activate
(weather) PS D:\learning\mcp\weather> 

如果出现下面错误,则表示没有权限,可以设置下权限

PS D:\learning\mcp\weather> Set-ExecutionPolicy RemoteSigned -Scope CurrentUser
image.png

执行脚本:

uv run client1.py

效果如下:


image.png

3. MCP客户端接入在线大模型

现在我们尝试在客户端中接入在线大模型进行对话。由于OpenAI和DeepSeek调用方法几乎一样,所以这个client可以同时支持对接GPT模型和DeepSeek模型。

3.1 新增依赖

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

uv add mcp openai python-dotenv
(weather) PS D:\learning\mcp\weather> uv add mcp openai python-dotenv          
Resolved 27 packages in 1.14s
Prepared 4 packages in 989ms
░░░░░░░░░░░░░░░░░░░░ [0/4] Installing wheels...                                                                                 
warning: Failed to hardlink files; falling back to full copy. This may lead to degraded performance.
         If the cache and target directories are on different filesystems, hardlinking may not be supported.
         If this is intentional, set `export UV_LINK_MODE=copy` or use `--link-mode=copy` to suppress this warning.
Installed 4 packages in 419ms
 + distro==1.9.0
 + jiter==0.9.0
 + openai==1.77.0
 + tqdm==4.67.1
(weather) PS D:\learning\mcp\weather>
3.2 创建.env文件

接下来创建.env文件,并在env文件中进行相应的大模型设置。

BASE_URL=https://api.siliconflow.cn/v1   # 大模型API地址
MODEL=deepseek-ai/DeepSeek-R1-Distill-Qwen-1.5B  #模型名称
OPENAI_API_KEY="xxxxxxx"    #你的API key
3.3 编写client代码

新建一个client2.py,把client1.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:
        """调用 OpenAI 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"⚠ 调用 OpenAI API 时出错: {str(e)}"

    async def chat_loop(self):
        """运行交互式聊天循环"""
        while True:
            try:
                query = input("\n你: ").strip()
                if query.lower() == 'quit':
                    break

                response = await self.process_query(query) # 发送用户输入到 OpenAI API
                print(f"\n🤖 OpenAI: {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()) 
3.4 运行代码
uv run client2.py

效果如下:


image.png
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容