基于MCP协议调用的大模型agent开发04

MCP客户端Client开发流程

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

uv安装流程

1、使用 pip 安装

pip install uv

2、使用 curl 直接安装

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

uv的基本用法介绍

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

示例

安装 Python 依赖

uv pip install numpy

与 pip install numpy 类似，但更快。

创建虚拟环境

uv venv mcpenv

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

激活虚拟环境

source mcpenv/bin/activate  # Linux/macOS
mcpenv\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

MCP是个让AI模型和外部工具协作的协议，就像给AI搭了个“工具箱”。但要管理这个工具箱里的东西，用传统的pip就像用手动拧螺丝，慢不说还容易出错。uv（虚拟环境管理工具）就厉害了，它就像个智能工具箱，能自动帮你搞定所有模块的安装和管理。

uv通过一个叫pyproject.toml的文件，把所有依赖都整得明明白白，就像给工具箱里的每个工具贴了标签，想找啥一目了然。而且uv的速度比pip快得多，安装东西就像开跑车，嗖一下就搞定了。对于MCP这种需要频繁折腾依赖的项目，uv简直是救星。

所以，MCP更推荐用uv来管理环境。接下来，咱们可以先搭个MCP客户端，就像先搭个积木架子，看看能不能跑起来。等客户端能正常工作了，再慢慢搭MCP服务器，就像给积木架子添砖加瓦。这样分阶段来，万一有问题也能一步步排查，不会一上来就被复杂性搞得晕头转向。

MCP极简客户端搭建流程

创建 MCP 客户端项目

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

建议在autodl上租用云算力进行测试部署

创建MCP客户端虚拟环境

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

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

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

# 安装 MCP SDK
uv add mcp

编写基础 MCP 客户端

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

写入代码

import asyncio
from mcp import ClientSession
from contextlib import AsyncExitStackclass 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" [Mock Response] 你说的是：{query}")
            except Exception as e:
                print(f" 发生错误: {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 退出程序。需要注意的是，此时客户端没有接入任何大模型，只会重复用户的输入。

代码解释：

MCP客户端基本代码结构

初始化 MCP 客户端

*提供一个命令行交互界面

模拟 MCP 服务器连接

支持用户输入查询并返回「模拟回复」

支持安全退出

首先是导入必要的库

import asyncio  # 让代码支持异步操作
from mcp import ClientSession  # MCP 客户端会话管理
from contextlib import AsyncExitStack  # 资源管理（确保客户端关闭时释放资源）

asyncio：Python 内置的异步编程库，让 MCP 可以非阻塞地执行任务（比如聊天、查询）。

mcp.ClientSession：用于管理 MCP 客户端会话（但目前我们先不连接 MCP 服务器）。

AsyncExitStack：自动管理资源，确保程序退出时正确关闭 MCP 连接。

然后创建 MCPClient 类

class MCPClient:
    def __init__(self):
        """初始化 MCP 客户端"""
        self.session = None  # 先不连接 MCP 服务器
        self.exit_stack = AsyncExitStack()  # 创建资源管理器

self.session = None：暂时不连接 MCP 服务器，后续可以修改来真正连接。

self.exit_stack = AsyncExitStack()：管理 MCP 客户端的资源，确保程序退出时可以正确释放资源。

模拟 MCP 服务器连接

async def connect_to_mock_server(self):
    """模拟 MCP 服务器的连接（暂不连接真实服务器）"""
    print(" MCP 客户端已初始化，但未连接到服务器")

这个函数不会真的连接 MCP 服务器，只是打印一条信息，表示客户端已经初始化。

async def：因为我们用的是 异步编程，所以需要用 async 关键字。

创建交互式聊天循环

async def chat_loop(self):
    """运行交互式聊天循环"""
    print("\nMCP 客户端已启动！输入 'quit' 退出")    while True:  # 无限循环，直到用户输入 'quit'
        try:
            query = input("\nQuery: ").strip()  # 让用户输入问题
            if query.lower() == 'quit':  # 如果用户输入 quit，退出循环
                break
            print(f" [Mock Response] 你说的是：{query}")  # 返回模拟回复
        except Exception as e:  # 发生错误时捕获异常
            print(f" 发生错误: {str(e)}")

while True：无限循环，让用户可以不断输入查询。

query = input("\nQuery: ").strip()：获取用户输入的查询。

if query.lower() == 'quit'：如果用户输入 quit，退出循环。

print(f" [Mock Response] 你说的是：{query}")：模拟 MCP 服务器的响应，暂时只是回显用户输入的内容。

async def cleanup(self):
    """清理资源"""
    await self.exit_stack.aclose()  # 关闭资源管理器

aclose() 确保程序退出时正确关闭 MCP 连接（尽管目前没有真正的连接）。

并定义**main() 主函数**

async def main():
    client = MCPClient()  # 创建 MCP 客户端
    try:
        await client.connect_to_mock_server()  # 连接（模拟）服务器
        await client.chat_loop()  # 进入聊天循环
    finally:
        await client.cleanup()  # 确保退出时清理资源

client = MCPClient()：创建一个 MCP 客户端实例。

await client.connect_to_mock_server()：初始化 MCP 客户端（暂不连接服务器）。

await client.chat_loop()：启动交互式聊天。

finally: 确保 不管程序是否异常退出，都会正确释放资源。

运行代码脚本

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

if __name__ == "__main__":：确保代码只能在 Python 直接运行时执行（而不是作为库导入时）

asyncio.run(main())：启动 main()，运行 MCP 客户端

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

运行 MCP 客户端

接下来尝试运行这个极简的MCP客户端：

uv run client.py

MCP客户端接入OpenAI、DeepSeek在线模型流程

接下来尝试在客户端中接入OpenAI和DeepSeek等在线模型进行对话。由于OpenAI和DeepSeek调用方法几乎完全一样，因此定义好一套的服务器client代码可以同时适用于GPT模型和DeepSeek模型。

新增依赖

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

uv add mcp openai python-dotenv

创建.env文件

接下来创建.env文件，并写入OpenAI的API-Key，以及反向代理地址。借助反向代理，国内可以无门槛直连OpenAI官方服务器，并调用官方API。

写入

BASE_URL="反向代理地址"
MODEL=gpt-4o
OPENAI_API_KEY="OpenAI-API-Key"

而如果是使用DeepSeek模型，则需要在.env中写入如下内容：

BASE_URL=https://api.deepseek.com
MODEL=deepseek-chat      
OPENAI_API_KEY="DeepSeek API-Key"

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")  # 读取 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:"""调用 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.contentexcept Exception as e:return f" 调用 OpenAI API 时出错: {str(e)}"async def chat_loop(self):"""运行交互式聊天循环"""print(" MCP 客户端已启动！输入 'quit' 退出")while True:try:
                query = input("\n你: ").strip()if query.lower() == 'quit':break                response = await self.process_query(query)  # 发送用户输入到 OpenAI APIprint(f" OpenAI: {response}")except Exception as e:print(f" 发生错误: {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())

运行client.py

然后即可输入如下命令开始运行对话客户端：

uv run client.py

clint.py代码解释

加载 OpenAI API Key

from dotenv import load_dotenv
import os# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 API Key
if not self.openai_api_key:raise ValueError(" 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")

load_dotenv()：自动加载 .env 文件，避免在代码中直接暴露 API Key。

os.getenv("OPENAI_API_KEY")：从环境变量中读取 OPENAI_API_KEY。

raise ValueError(...)：如果 API Key 为空，则抛出错误，提醒用户必须配置 API Key。

创建 .env 文件

touch .env

在 .env 文件中添加 API Key

OPENAI_API_KEY=你的OpenAI API Key

发送用户输入到 OpenAI API

async def process_query(self, query: str) -> str:
    """调用 OpenAI API 处理用户查询"""
    messages = [
        {"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},
        {"role": "user", "content": query}
    ]    try:
        # 调用 OpenAI GPT-4 API
        response = await asyncio.get_event_loop().run_in_executor(
            None,
            lambda: openai.ChatCompletion.create(
                model="gpt-4",
                messages=messages,
                max_tokens=1000,
                temperature=0.7
            )
        )
        return response["choices"][0]["message"]["content"].strip()
    except Exception as e:
        return f" 调用 OpenAI API 时出错: {str(e)}"

messages：创建对话上下文，让 OpenAI 知道如何回答问题：

system 角色：设定 AI 角色（如“你是一个智能助手”）。

user 角色：存储用户输入。

openai.ChatCompletion.create(...)

model="gpt-4"：使用 OpenAI 的 GPT-4 进行对话。

messages=messages：提供聊天记录，让 AI 生成回答。

max_tokens=1000：限制 AI 生成的最大字数。

temperature=0.7：控制 AI 回答的随机性（越高越随机）。

run_in_executor(...)：

因为 OpenAI API 是同步的，但我们用的是异步代码

这里用 asyncio.get_event_loop().run_in_executor(...) 将 OpenAI API 变成异步任务，防止程序卡顿。

交互式聊天

async def chat_loop(self):"""运行交互式聊天循环"""print(" MCP 客户端已启动！输入 'quit' 退出")while True:try:
            query = input("\n你: ").strip()if query.lower() == 'quit':break            response = await self.process_query(query)  # 发送用户输入到 OpenAI APIprint(f" OpenAI: {response}")except Exception as e:print(f" 发生错误: {str(e)}")

输入查询 query = input("\n你: ").strip()，支持多轮对话。

调用 process_query()，将用户输入发送到 OpenAI API 并获取回复。

显示 OpenAI 生成的回复：print(f" OpenAI: {response}")

用户输入 quit 退出。

需要注意的是，由于MCP的client SDK主要规定了client和server之间的通信方法，因此在没有创建server之前，一个单纯对话的client甚至不需要用到mcp功能。但本段代码的学习仍是有必要的，为了熟悉各类大模型本地调用对话流程。而后我们只需要围绕上述代码稍作修改，即可调用外部的server。