当前位置： 首页 > news >正文

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

news 来源：原创 2025/6/23 14:59:41

目录

MCP客户端Client开发流程

uv工具

uv安装流程

uv的基本用法介绍

MCP极简客户端搭建流程

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

参考来源及学习推荐：

Anthropic MCP发布通告：https://www.anthropic.com/news/model-context-protocol

MCP GitHub主页：https://github.com/modelcontextprotocol

本系列：

https://blog.csdn.net/yuange1666/article/details/147095760?spm=1001.2014.3001.5502
https://blog.csdn.net/yuange1666/article/details/147095853?spm=1001.2014.3001.5502
https://blog.csdn.net/yuange1666/article/details/147127585?spm=1001.2014.3001.5501
https://blog.csdn.net/yuange1666/article/details/147127981?spm=1001.2014.3001.5501

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 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" [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")  # 读取 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):
        """运行交互式聊天循环"""
        print(" MCP 客户端已启动！输入 'quit' 退出")

        while True:
            try:
                query = input("\n你: ").strip()
                if query.lower() == 'quit':
                    break
                
                response = await self.process_query(query)  # 发送用户输入到 OpenAI API
                print(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 API
            print(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。

相关文章：

  • 【GDB】调试程序的基本命令和用法（Qt程序为例）
  • Harmony鸿蒙应用开发-录音保存并播放音频
  • 《穿透表象，洞察分布式软总线“无形”之奥秘》
  • GreenStableYolo：多目标优化文本到图像生成的推理时间与图像质量
  • 操作主机的管理
  • 基于SpringBoot的在线政务服务中心(源码+数据库+万字文档+ppt)
  • 区块链 数据共享 搭建尝试
  • 【FreeRTOS】什么是钩子 函数
  • HTTP 协议详解
  • Java基础关键_035_Lambda 表达式
  • 【特权FPGA】之AT24C02 IIC实现
  • 光流 ｜ 近十年（2015–2025年）光流算法研究的代表性文献综述及光流开源项目
  • Langchat平台知识库测试
  • MySQL--基础知识点--81.2--EVENT
  • Spark core编程之RDD 转换算子分类
  • java并发编程面试之ThreadLocal深度解析
  • python多线程+异步编程让你的程序运行更快
  • 蓝桥杯 分解质因数（唯一分解定理）
  • 【专题】贪心算法
  • [C语言]gets和fgets函数区别及详解
  • 阿里云建站后台建站/网站seo入门基础教程
  • 昆山网站优化建设/新闻联播俄罗斯与乌克兰
  • 云南建设工程有限公司/科学新概念seo外链平台
  • 毕设代做有哪些靠谱网站/seo搜索引擎优化工作内容
  • 设计网站推荐 猪/各大搜索引擎收录入口
  • 怎样做网站平台赚钱/网络营销品牌案例