网站导航
> 技术文档 > Python----MCP（MCP 简介、uv工具、创建MCP流程、MCP客户端接入Qwen、MCP客户端接入vLLM）

Python----MCP（MCP 简介、uv工具、创建MCP流程、MCP客户端接入Qwen、MCP客户端接入vLLM）

网友: 技术文档 2025-08-08 12:33:42

一、MCP 简介

        Model Context Protocol (MCP)是一个开放协议，它为应用程序向 LLM 提供上下文的方式进行了标准化。你可以将 MCP 想象成 AI 应用程序的 USB-C 接口。就像 USB-C 为设备连接各种外设和配件提供了标准化的方式一样，MCP 为 AI 模型连接各种数据源和工具提供了标准化的接口。

        MCP 帮助你在 LLM 的基础上构建代理（agents）和复杂的工作流。LLM 经常需要与数据和工具集成，而 MCP 提供了：

  • 持续增长的预构建集成列表，LLM 可直接使用

  • 灵活切换不同的 LLM 提供商和厂商

  • 在你的基础设施内安全地处理数据的最佳实践

 MCP解决的最大痛点，就是Agent开发中调用外部工具的技术门槛过高的问题。

  • MCP Hosts: 如 Claude Desktop、IDE 或 AI 工具，希望通过 MCP 访问数据的程序
  • MCP Clients: 维护与服务器一对一连接的协议客户端
  • MCP Servers: 轻量级程序，通过标准的 Model Context Protocol 提供特定能力
  • 本地数据源: MCP 服务器可安全访问的计算机文件、数据库和服务
  • 远程服务: MCP 服务器可连接的互联网上的外部系统（如通过 APIs）

​

二、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。

三、创建MCP流程

3.1、安装uv

pip install uv

3.2、安装mcp

pip install mcp

3.3、为项目创建一个新目录（以weather为例）

uv init weather
cd weather

3.4、创建虚拟环境并激活

uv venv
source .venv/bin/activate

3.5、安装依赖项

uv add \"mcp[cli]\" httpx

3.6、创建服务器文件

touch weather.py

 3.7、编写基础 MCP 客户端

        创建 client.py

import asyncio # Python 内置的异步编程库，让 MCP 可以非阻塞地执行任务from mcp import ClientSession # 管理 MCP 客户端会话from contextlib import AsyncExitStack # 自动管理资源，确保程序退出时正确关闭 MCP 连接class MCPClient: def __init__(self): \"\"\"初始化 MCP 客户端\"\"\" self.session = None # 暂时不连接 MCP 服务器，后续可以修改来真正连接。 self.exit_stack = AsyncExitStack() #管理 MCP 客户端的资源，确保程序退出时可以正确释放资源。# - 这个函数不会真的连接 MCP 服务器，只是打印一条信息，表示客户端已经初始化。 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()# 确保程序退出时正确关闭 MCP 连接async def main(): client = MCPClient()# 创建一个 MCP 客户端实例 try: await client.connect_to_mock_server()#初始化 MCP 客户端（暂不连接服务器） await client.chat_loop()#启动交互式聊天。 finally: await client.cleanup()if __name__ == \"__main__\": asyncio.run(main())

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

四、MCP客户端接入Qwen

        创建.env文件

BASE_URL=https://api.siliconflow.cn/v1MODEL=Qwen/Qwen3-8B OPENAI_API_KEY=\"sk-kjutyipwobagmkhdzgxszqusfueaquhhrzvyuasyedwqfzkbrrv\"#替换成自己的api密钥，使用的是硅基流动

        客户端代码

import asyncioimport osfrom openai import OpenAIfrom dotenv import load_dotenvfrom 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(\"\\n🤖 MCP 客户端已启动！输入 \'quit\' 退出\") 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())

 五、MCP客户端接入vLLM

启动vllm

python -m vllm.entrypoints.openai.api_server --port 10222 --model /home/AI_big_model/models/Qwen/Qwen2.5-7B-Instruct --served-model-name Qwen2.5-7B-Instruct

.env

BASE_URL=http://127.0.0.1:10222/v1MODEL=Qwen2.5-7B-InstructAPI_KEY=EMPTY

客户端不变

import asyncioimport osfrom openai import OpenAIfrom dotenv import load_dotenvfrom 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\") # 读取 BASE YRL self.model = os.getenv(\"MODEL\") # 读取 model if not self.api_key: raise ValueError(\"❌ 未找到 API Key，请在 .env 文件中设置 api_key\")  self.client = OpenAI(api_key=self.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(\"\\n🤖 MCP 客户端已启动！输入 \'quit\' 退出\") 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())

网络标签：

相关问题