一、MCP基本概念

MCP（Model Context Protocol，模型上下文协议）是由Anthropic主导发布的开放标准协议，旨在让大型语言模型（LLM）与外部数据源和工具无缝集成。它类似于AI领域的“USB-C”接口，为AI模型提供了一个标准化的外部数据交互接口。

1.1 主要用途

1. 连接外部数据源和工具：MCP允许LLM访问数据库、API、文件系统、搜索引擎等外部数据源，以及执行各种操作的工具。例如，通过MCP，AI可以读取本地文件、查询数据库、调用API等。
2. 增强AI功能：通过集成外部工具和服务，MCP扩展了AI应用的功能范围，使其能够执行更复杂的任务。比如，AI可以利用MCP服务器提供的功能，实现文件编辑、数据分析、图像生成等操作。
3. 简化开发流程：MCP提供了一种统一的通信协议，减少了为每个数据源或工具单独开发接口的需求。开发者可以利用现有的开源MCP服务，快速集成各种功能，无需重复造轮子。
4. 保障数据安全：MCP定义了明确的接口和访问控制，确保了数据传输的安全性。用户可以自行决定哪些数据需要传输，敏感数据可以保留在本地，不上传到云端。
5. 支持多轮对话：MCP能够动态管理LLM的对话上下文，确保多轮对话的连贯性和一致性。
6. 促进AI生态发展：MCP为服务提供商提供了一个开放的标准，有助于构建一个更加开放和协作的AI应用生态系统。例如，出现了MCP做的“App Store” Fleur，以及MCP导航网站mcp.so，收录了近3000个MCP Server。

1.2 应用场景

• 增强型问答系统：通过集成外部数据源，提供实时、准确的答案。
• 智能助手：通过集成工具和服务，执行复杂任务，如预订、计算、搜索等。
• 知识管理：通过集成文档库和数据库，提供专业领域的知识支持。
• 多轮对话：通过上下文管理，实现连贯的多轮对话。

二、开发工具uv

在MCP（模型上下文协议）的开发和使用中，uv工具是一个重要的辅助工具，主要用于运行和管理基于Python的MCP服务器，类似于pip和conda，但是uv更快，更高效。

2.1 uv工具介绍

uv是一个用Rust编写的Python包和项目管理器，具有以下特点：

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

2.2 uv工具安装流程

uv可以通过以下几种方式安装：

• 使用pip安装

 pip install uv

你也可以使用curl安装，安装命令如下

• 使用curl直接安装

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

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

下面我就直接切入主题，来教大家如何结合uv命令构建MCP客户端。

三、MCP客户端搭建示例

3.1 初始化MCP客户端项目

#创建MCP客户端的目录
uv init mcp-client-test
cd mcp-client-test

执行完uv init mcp-client-test命令后，会自动创建mcp客户端的项目结构，项目结构如下：

我详细解释一下以上图片中三个文件及其用途如下：

• main.py

   - 这是项目的主程序文件，通常包含项目的入口点和主要逻辑。
   - 在MCP项目中，main.py可能包含与MCP服务器交互的代码，例如定义如何处理请求、如何调用外部工具或服务等。

• pyproject.toml

   - 这是一个配置文件，用于定义Python项目的构建系统和依赖关系。
   - 在MCP项目中，pyproject.toml可能包含项目的元数据（如项目名称、版本、作者等），以及项目所需的Python依赖包列表。
   - 它还可能包含构建系统的配置，如如何构建项目、如何安装依赖等。
   - 使用pyproject.toml可以方便地管理项目的依赖和构建过程，提高项目的可移植性和可维护性。

具体内容如下：

• README.md

   - 这是一个Markdown格式的文档文件，通常用于提供项目的概述、使用说明、安装指南等信息。
   - 在MCP项目中，README.md可能包含以下内容：
     - 项目简介：简要介绍项目的目的和功能。
     - 安装指南：说明如何安装和配置项目。
     - 使用说明：提供如何使用项目的详细步骤和示例代码。
     - 配置选项：列出项目支持的配置选项及其说明。
     - 贡献指南：说明如何为项目做出贡献，包括提交代码、报告问题等。
   - README.md是项目的第一印象，对于吸引用户和开发者非常重要。

这些文件共同构成了一个典型的MCP项目结构，分别承担着不同的角色和职责，确保项目的可读性、可维护性和可扩展性。

3.2 创建MCP客户端虚拟环境

创建虚拟环境，你可以使用uv venv命令，有一点需要注意，相比pip，uv会自动识别当前项目主目录并创建虚拟环境，所以你只要执行uv venv命令即可。

图片中显示的解释器路径是 /Library/Developer/CommandLineTools/usr/bin/python3，这是因为macOS系统自带了一个Python解释器，位于 /Library/Developer/CommandLineTools/usr/bin/python3。当你在终端中输入 python3 时，默认会调用这个解释器。所以，会导致后面在安装MCP SDK库时报错，因为macOS系统自带的python解释器版本太低了。

然后，激活虚拟环境，执行如下：

出现红框内容，表示已经成功激活虚拟环境，并且在mcp-client-test的虚拟环境下了。

3.3 安装MCP SDK的库

安装MCP SDK库，执行如下命令：

uv add mcp

执行后，如果你也报了如下错误，说明你当前安装的python版本级别太低，因为请求的Python版本（>=3.9）不满足mcp依赖的Python版本要求（>=3.10）。这意味着所有版本的mcp都不能使用，因为它们只支持Python 3.10及以上版本。

所以，你可以先检查你的python版本，如果你的电脑中已经安装了更高阶版本的Python解释器，你就设定环境变量PYTHON，后续创建uv虚拟环境中，将使用默认Python解释器来创建。检查当前虚拟环境下的python版本：

python --version

因为我已经安装了conada，所以设置PYTHON的环境变量。

3.4 编写MCP客户端程序

创建一个client.py的文件，实现MCP客户端的功能，具体代码如下，每一行代码，我都给了详细的注释：

# 导入必要的模块
import asyncio  # 导入异步 IO 模块，用于处理异步操作
import logging  # 导入日志模块，用于记录程序运行状态
from typing import Optional  # 导入类型提示模块中的 Optional 类型

class MCPClient:  # 定义 MCP 客户端类
    def __init__(self, host: str = "localhost", port: int = 8080):  # 初始化方法，设置默认主机和端口
        """初始化 MCP 客户端
        
        Args:
            host: 服务器主机地址
            port: 服务器端口
        """
        self.host = host  # 存储服务器主机地址
        self.port = port  # 存储服务器端口
        self.connected = False  # 初始化连接状态为未连接
        self.reader: Optional[asyncio.StreamReader] = None  # 初始化读取器为 None
        self.writer: Optional[asyncio.StreamWriter] = None  # 初始化写入器为 None
        self.logger = logging.getLogger(__name__)  # 创建日志记录器
        
    async def connect(self) -> bool:  # 异步连接方法，返回布尔值表示连接是否成功
        """异步连接到 MCP 服务器
        
        Returns:
            bool: 连接是否成功
        """
        try:  # 尝试建立连接
            self.reader, self.writer = await asyncio.open_connection(  # 异步打开连接
                self.host, self.port  # 使用存储的主机和端口
            )
            self.connected = True  # 设置连接状态为已连接
            self.logger.info(f"已连接到服务器 {self.host}:{self.port}")  # 记录连接成功信息
            return True  # 返回连接成功
        except Exception as e:  # 捕获连接过程中的异常
            self.logger.error(f"连接服务器失败: {e}")  # 记录连接失败信息
            return False  # 返回连接失败
            
    async def disconnect(self):  # 异步断开连接方法
        """断开与服务器的连接"""
        if self.writer:  # 如果写入器存在
            self.writer.close()  # 关闭写入器
            await self.writer.wait_closed()  # 等待写入器完全关闭
            self.connected = False  # 设置连接状态为未连接
            self.logger.info("已断开与服务器的连接")  # 记录断开连接信息
            
    async def send_message(self, message: str):  # 异步发送消息方法
        """发送消息到服务器
        
        Args:
            message: 要发送的消息
        """
        if not self.connected or not self.writer:  # 如果未连接或写入器不存在
            self.logger.error("未连接到服务器")  # 记录错误信息
            return  # 直接返回
            
        try:  # 尝试发送消息
            self.writer.write(message.encode())  # 将消息编码并写入
            await self.writer.drain()  # 等待数据发送完成
            self.logger.debug(f"已发送消息: {message}")  # 记录发送消息信息
        except Exception as e:  # 捕获发送过程中的异常
            self.logger.error(f"发送消息失败: {e}")  # 记录发送失败信息
            
    async def receive_message(self) -> Optional[str]:  # 异步接收消息方法，返回可选字符串
        """从服务器接收消息
        
        Returns:
            Optional[str]: 接收到的消息，如果连接断开则返回 None
        """
        if not self.connected or not self.reader:  # 如果未连接或读取器不存在
            return None  # 返回 None
            
        try:  # 尝试接收消息
            data = await self.reader.read(1024)  # 异步读取最多 1024 字节的数据
            if not data:  # 如果没有接收到数据
                return None  # 返回 None
            return data.decode()  # 解码并返回接收到的数据
        except Exception as e:  # 捕获接收过程中的异常
            self.logger.error(f"接收消息失败: {e}")  # 记录接收失败信息
            return None  # 返回 None
            
    async def chat_loop(self):  # 异步聊天循环方法
        """运行交互式聊天循环"""
        while self.connected:  # 当连接状态为已连接时循环
            try:  # 尝试执行聊天循环
                # 接收用户输入
                message = await asyncio.get_event_loop().run_in_executor(  # 在事件循环中运行阻塞的输入操作
                    None, input, "请输入消息 (输入 'quit' 退出): "  # 使用 input 函数获取用户输入
                )
                
                if message.lower() == 'quit':  # 如果用户输入 'quit'
                    break  # 退出循环
                    
                # 发送消息
                await this.send_message(message)  # 异步发送用户输入的消息
                
                # 接收服务器响应
                response = await this.receive_message()  # 异步接收服务器响应
                if response:  # 如果接收到响应
                    print(f"服务器响应: {response}")  # 打印服务器响应
                    
            except Exception as e:  # 捕获聊天循环中的异常
                self.logger.error(f"聊天循环出错: {e}")  # 记录错误信息
                break  # 退出循环
                
    async def cleanup(self):  # 异步清理资源方法
        """清理资源"""
        await this.disconnect()  # 异步断开连接
        self.logger.info("资源清理完成")  # 记录资源清理完成信息

async def main():  # 异步主函数
    """主函数"""
    # 配置日志
    logging.basicConfig(  # 配置日志记录
        level=logging.INFO,  # 设置日志级别为 INFO
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'  # 设置日志格式
    )
    
    # 创建客户端实例
    client = MCPClient()  # 创建 MCP 客户端实例
    
    try:  # 尝试执行主要操作
        # 连接到服务器
        if await client.connect():  # 如果连接成功
            # 运行聊天循环
            await client.chat_loop()  # 运行聊天循环
    except KeyboardInterrupt:  # 捕获键盘中断异常
        logging.info("用户中断程序")  # 记录用户中断信息
    finally:  # 无论是否发生异常，都执行清理操作
        # 清理资源
        await client.cleanup()  # 清理资源

if __name__ == "__main__":  # 如果直接运行此文件
    # 运行主函数
    asyncio.run(main())  # 运行异步主函数