【2025】🔥MCP是什么？应该怎么使用？🚀本期主要讲MCP客户端

什么是MCP？

MCP （Model Context Protocol，即模型上下文协议），定义了应用程序和 AI 模型之间交换上下文信息的方式。这使得开发者能够以一致的方式将各种数据源、工具和功能连接到 AI 模型（一个中间协议层），就像 USB-C 让不同设备能够通过相同的接口连接一样.。MCP 的目标是创建一个通用标准，使 AI 应用程序的开发和集成变得更加简单和统一。

一、uv安装

1.1 安装uv的方法及途径

方法1：使用pip安装（适用已安装pip的系统：Windows）

pip install uv

方法2：使用curl安装（适用未安装pip的系统：macos&Linux）

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

其余安装方法请参考：https://docs.astral.sh/uv/getting-started/installation/#scoop

1.2 创建虚拟环境

uv venv 

1.3 激活虚拟环境

source .venv/bin/activate // Linux/macos
.venv\Scripts\activate // Windows

二、创建MCP客户端项目

2.1 初始化MCP客户端

uv init mcp-client
cd mcp-client

2.2 创建MCP客户端虚拟环境

创建虚拟环境并激活

uv venv 
.venv\Scripts\activate

安装MCP SDK

uv add mcp

2.3 编写基础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"\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())

【注意】：如果提示没有依赖请自行下载

2.4 运行MCP客户端

uv run client.py

三、MCP客户端接入OpenAI、DeepSeek等模型

3.1 新增依赖

由于绝大部分模型均支持使用OpenAI 兼容的 API 格式，所以直接使用openai SDK调用模型api即可。

uv add mcp openai python-dotenv

3.2 创建.env文件

此文件是存放大模型的API-Key，以及反向代理地址等信息.设置正确的代理地址即可无门槛访问Chat-GPT等国外大模型

BASE_URL="反向代理地址" //根据不同的大模型使用不同的base_url
MODEL=模型名称 
OPENAI_API_KEY="API-Key" //申请api-key后,填入即可

【切记】复制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.api_key = os.getenv("API_KEY") # 读取API Key
        self.base_url = os.getenv("BASE_URL")
        self.model=os.getenv("MODEL")

        if not self.api_key:
            raise ValueError("未找到API key,请在.env文件中添加正确的key！")

        self.client = OpenAI(api_key=self.api_key,base_url=self.base_url)

    async def process_query(self,query:str)->str:
        """调用deepseek V3处理用户查询"""
        messages = [
            {"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},
            {"role": "user", "content": query},
        ]

        try:
            #调用deepseek 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时出错：{str(e)}"



    async def chat_loop(self):
        """运行交互式聊天循环"""
        print("\n MCP 客户端已启动！输入 'quit' 退出！")

        while True:
            try:
                query = input("\n 你: ").strip()
                if query.lower() == "quit":
                    break

                response = await self.process_query(query)

                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())

3.4 接入本地大模型

接入本地大模型只需要修改.env文件内容即可，这里采用ollama下载的qwen2.5:1.5b模型来展示

BASE_URL=http://localhost:11434/v1/ //默认地址
MODEL=qwen2.5:1.5b //修改为自己本地下载的模型名称
API_KEY=ollama //默认本地不使用API_KEY,直接填入ollama即可