一、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 uv3.2、安装mcp
pip install mcp3.3、为项目创建一个新目录（以weather为例）
uv init weathercd weather3.4、创建虚拟环境并激活
uv venvsource .venv/bin/activate3.5、安装依赖项
uv add "mcp[cli]" httpx3.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/v1
MODEL=Qwen/Qwen3-8B
OPENAI_API_KEY="sk-kjutyipwobagmkhdzgxszqusfueaquhhrzvyuasyedwqfzkbrrv"
#替换成自己的api密钥，使用的是硅基流动客户端代码
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("\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/v1
MODEL=Qwen2.5-7B-Instruct
API_KEY=EMPTY客户端不变
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") # 读取 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())