• 👏作者簡介：大家好，我是愛吃芝士的土豆倪，24屆校招生Java選手，很高興認識大家
• 📕系列專欄：Spring原理、JUC原理、Kafka原理、分布式技術原理、數據庫技術、JVM原理、AI應用
• 🔥如果感覺博主的文章還不錯的話，請👍三連支持👍一下博主哦
• 🍂博主正在努力完成2025計劃中：逆水行舟，不進則退
• 📝聯系方式：nhs19990716，加我進群，大家一起學習，一起進步，一起對抗互聯網寒冬👀

簡述

今天簡單聊聊mcp-client客戶端，并接入deepseek 模型，調用我們自己的mcp server，本篇是基于之前開發好的mcp-server進行的。并且使用python去實現mcp client的客戶端。

什么是mcp的客戶端呢？mcp的客戶端通俗易懂的來講，就像是一個翻譯官，它具備對英語或者漢語或其他語種的翻譯語種，語種可以理解為一個具體的ai模型，它可以幫助我們和mcp服務進行通信，就比如我們有一個專門用于天氣查詢的mcp服務，但是這個服務只會根據它熟悉的格式或者語言去查詢天氣，那么mcp客戶端就是把我們的自然語言，也就是普通的描述，比如你幫我查詢天氣，然后mcp客戶端會把這個普通語言翻譯成天氣查詢mcp服務理解的格式或者語言，最終將天氣查詢mcp服務返回的數據再翻譯成普通語言告訴我們，那么這個過程其實就是mcp客戶端 加 模型一起完成的，可以把mcp客戶端想象成翻譯官，那么它所掌握的某個語言，比如說漢語，英語其實就是模型。

環境準備

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

# 創建目錄
uv init mcp-clientcd mcp-client# 創建虛擬環境
uv env# 激活虛擬環境
source .venv/bin/activate# 安裝相關的依賴
uv add mcp anthropic python-dotenv# 創建一個程序入口文件
touch client.py

# 創建環境變量文件，用于存儲模型的api key
touch .envAPI_KEY = 去deepseek官網申請即可
BASE_URL = 去官網查即可
MODEL_NAME = 去官網搜模型名稱 使用的是deepseek v3，因為官網的文檔里明確說了r1不支持function call

創建client客戶端

初始化

# 用于導入異步IO庫，用于支持異步編程
import asyncio 
# 用于導入JSON庫，用于處理JSON數據
import json
# 用于處理命令行參數
import sys 
# 用于類型提示功能
from typing import Optional 
# 異步資源管理器，用于管理多個異步資源
from contextlib import AsyncExitStack # MCP 客戶端相關導入
# 導入 MCP 客戶端會話和標準輸入輸出服務器參數
from mcp import ClientSession, StdioServerParameters 
# 導入標準輸入輸出客戶端通信模塊
from mcp.client.stdio import stdio_client 
# Openai SDK
from openai import OpenAI # 環境變量加載相關
# 導入環境變量加載工具
from dotenv import load_dotenv 
# 用于獲取環境變量值
import os # 加載 .env 文件中的環境變量
load_dotenv()  # 定義 MCP 客戶端類
class DeepSeekMCPClient:"""使用 DeepSeek V3 API 的 MCP 客戶端類處理 MCP 服務器連接和 DeepSeek V3 API 的交互這個類就像是一個翻譯官，一方面與MCP服務器進行通信，另一方面與DeepSeek API進行通信，幫助用戶通過自然語言來使用各種強大的工具"""def __init__(self):""" 初始化MCP客戶端的各項屬性主要設置了三個重要組件：- session: 用于與MCP服務器通信的會話- exit_stack: 用于管理異步資源的上下文管理器，確保資源正確釋放，那么在與 MCP 服務通信時，它會負責接收和發送通信數據- llm_client: DeepSeek API 的客戶端，使用 OpenAI 的 SDK"""# MCP 客戶端會話，初始值為 Noneself.session: Optional[ClientSession] = None# 創建異步資源管理器，用于管理多個異步資源self.exit_stack = AsyncExitStack()# 初始化 DeepSeek API 客戶端self.llm_client = OpenAI(api_key=os.getenv("API_KEY"), # 從環境變量中獲取 API 密鑰base_url=os.getenv("BASE_URL") # 從環境變量中獲取 API 基礎 URL)# 從環境變量獲取模型名稱self.model = os.getenv("MODEL")

封裝連接mcp-server函數

async def connect_to_server(self, server_script_path: str):"""連接到MCP服務這個函數就像是撥通電話，建立與 MCP 服務器的連接，它會根據服務腳本的類型(Python 或 JavaScript)選擇正確的命令啟動服務器，然后與之建立通信。參數:server_script_path: MCP 服務腳本路徑，支持 Python(.py) 或 Node.js(.js) 文件異常:ValueError: 如果服務器腳本不是.py或.js文件"""# 檢查腳本類型is_python = server_script_path.endswith('.py') # 判斷是否是 Python 腳本is_js = server_script_path.endswith('.js') # 判斷是否是 JavaScript 腳本if not (is_python or is_js): # 如果腳本類型不是 Python 或 JavaScript，則拋出異常raise ValueError("服務器腳本必須是 .py 或 .js 文件")# 根據腳本類型選擇正確的運行命令command = "python" if is_python else "node" # Python 使用 python 運行，JavaScript 使用 node 運行# 設置服務器啟動參數，那么 server_params 最終會生成類似于 Python xxx.py 這種運行命令server_params = StdioServerParameters(command=command, # 要執行的命令(python 或 node)args=[server_script_path], # 要執行的命令的參數(腳本路徑)env=None # 環境變量, 使用 None 表示繼承當前環境變量)# 創建標準輸入輸出通信信道stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))# 解構對象中的讀寫通信，分別用于向MCP服務接收和發送數據self.stdio, self.write = stdio_transport# 創建MCP客戶端會話self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))# 初始化MCP客戶端會話await self.session.initialize() # 初始化會話，準備好與MCP服務進行通信# 列出可用的工具# 獲取 MCP 服務提供的工具列表response = await self.session.list_tools()# 獲取工具列表tools = response.tools# 打印工具列表print("\n已連接到MCP服務，可用的工具列表:", [tool.name for tool in tools])

這個函數的作用就是通過命令運行一個mcp-server，然后使用mcp-client客戶端連接這個mcp-server，要保持一個會話狀態，然后我們就可以通過這個session去請求到這個mcp-server了，然后去獲取到這個mcp-server所提供的工具tools了

處理用戶查詢的核心函數

async def process_query(self, query: str) -> str:"""處理用戶查詢，根據查詢參數使用DeepSeek V3和MCP工具這個函數就是整個系統的核心，他就像一個指揮官，它接收用戶的問題，然后再把問題交給AI模型然后模型決定使用哪些工具，然后告訴 MCP Client 去調用這些工具。然后獲取到工具結果再返回給模型，模型根據工具結果生成最終的回答。整個過程就像是：用戶提問->模型判斷->MCP Client 使用工具->返回工具結果->模型根據工具結果生成回答參數：query: 用戶的問題返回：str: 處理后的最終響應的文本"""# 創建消息列表，用于存儲用戶的問題和模型的回答messages = [{# 系統角色，用于設定AI的行為準則"role": "system", "content": "你是一個專業的助手，可以通過調用合適的工具來幫助用戶解決問題，請根據用戶的需求選擇最合適的工具。"},{# 用戶角色，表示這是用戶發送的消息"role": "user", "content": query}]# 請求 MCP 服務獲取服務提供的工具列表response = await self.session.list_tools()# 獲取工具列表tools = response.tools# 構建工具信息數組，我們需要把工具信息轉換成 DeepSeek API 需要的格式available_tools = [{"type": "function", # 工具類型，表示這是一個函數工具"function": { # 工具的詳細定義"name": tool.name, # 工具名稱"description": tool.description, # 工具描述"parameters": tool.inputSchema # 工具參數}} for tool in tools]# 打印可用工具信息，便于調試print(f"當前 MCP 服務所有工具列表: {available_tools}\n--------------------------------\n")

調用deepseek返回可用工具

# 調用 DeepSeek API，發送用戶查詢和可用工具信息，
# 告訴 DeepSeek API 根據用戶提問你可以使用哪些工具，最終返回可調用的工具
response = self.llm_client.chat.completions.create(# 指定的模型名稱model=self.model, # 消息歷史（系統提示和用戶問題）messages=messages, # 可用的工具列表tools=available_tools if available_tools else None, # 溫度參數，控制響應的隨機性(0.5是中等隨機性)temperature=0.5, # 最大生成令牌數，限制響應長度max_tokens=4096 
)
# 打印模型響應，便于調試
print(f"DeepSeek API 響應: {response}\n--------------------------------\n")# 獲取模型的回復，包含 role(消息發送者) 和 
# content(消息內容) 以及 tool_calls(工具調用請求)
reply = response.choices[0].message # 獲取模型的回答# 打印模型的回答
print(f"DeepSekk 初始回復: {reply}\n--------------------------------\n")# 初始化最終文本結果列表
final_text = []# 將模型回復添加到歷史消息中，用于維護完整的對話歷史
# 這一步非常重要，確保模型 記得 自己之前決定使用什么工具，
# 即使模型沒有請求調用工具，也要保持對話連貫性。
messages.append(reply)

檢查模型響應中是否包含工具調用請求

# 檢查模型響應中是否包含工具調用請求，如果用戶的問題涉及到使用工具，
# 那就會包含 tool_calls 字段,否則就沒有
if hasattr(reply, "tool_calls") and reply.tool_calls:# 遍歷所有工具調用請求for tool_call in reply.tool_calls:# 獲取工具名稱tool_name = tool_call.function.name# 獲取工具參數try:# 嘗試將工具的參數從 JSON 字符串解析為 Python 字典tool_args = json.loads(tool_call.function.arguments)except json.JSONDecodeError:tool_args = {}# 打印工具調用信息，便于調試print(f"準備調用工具: {tool_name} 參數: {tool_args}\n--------------------------------\n")# 異步調用 MCP 服務上的工具，傳入工具名稱和函數參數，返回工具函數執行結果result = await self.session.call_tool(tool_name, tool_args)# 打印工具執行結果，便于調試print(f"工具 {tool_name} 執行結果: {result}\n--------------------------------\n")# 將工具調用信息添加到最終輸出文本中，便于用戶了解執行過程final_text.append(f"調用工具: {tool_name}, 參數: {tool_args}\n")# 確保工具結果是字符串格式tool_result_content = result.contentif isinstance(tool_result_content, list):# 如果工具結果是列表，則將列表中的每個元素轉換為字符串并添加到最終文本中text_content = ""for item in tool_result_content:if hasattr(item, 'text'):text_content += item.texttool_result_content = text_contentelif not isinstance(tool_result_content, str):# 如果不是字符串，則轉換為字符串tool_result_content = str(tool_result_content)# 打印工具返回結果print(f"工具返回結果(格式化后): {tool_result_content}\n--------------------------------\n")# 將工具調用結果添加到歷史消息中，保證與模型會話的連貫性tool_message = {# 工具角色，表示這是工具返回的結果"role": "tool", # 工具調用ID"tool_call_id": tool_call.id,# 工具返回的結果"content": tool_result_content, }# 打印消息內容print(f"添加到歷史消息中的工具消息: {tool_message}\n--------------------------------\n")# 添加到歷史消息中messages.append(tool_message)# 再次調用 DeepSeek API，讓模型根據工具結果生成最終的回答try:print("正在請求 DeepSeek API 生成最終回答...")# 發送包含工具調用和結果的完整消息歷史final_response = self.llm_client.chat.completions.create(model=self.model, # 指定的模型名稱messages=messages, # 消息歷史（系統提示和用戶問題）temperature=0.5, # 溫度參數，控制響應的隨機性(0.5是中等隨機性)max_tokens=4096 # 最大生成令牌數，限制響應長度)# 添加 DeepSeek 對工具結果的解釋然后到最終輸出final_content = "DeepSeek回答：" + final_response.choices[0].message.contentif final_content:# 如果模型生成了對工具結果的解釋，就將其添加到最終輸出數組中final_text.append(final_content)else:print("警告：DeepSeek API 沒有生成任何內容。")# 如果沒用內容，直接顯示工具結果final_text.append(f"工具調用結果：\n{tool_result_content}")except Exception as e:print(f"生成最終回復時出錯: {e}")final_text.append(f"工具返回結果：\n{tool_result_content}")
else:# 如果模型沒有請求調用工具，那么就直接返回模型的內容if reply.content:# 將模型的直接回復添加到最終輸出數組final_text.append(f"{reply.content}")else:# 如果模型沒有生成內容，則添加提示信息final_text.append("模型沒有生成有效回復。")# 我們把用戶的問題和MCP服務可用工具全部給到 DeepSeek，# DeepSeek 判斷出具體需要調用哪個工具，然后讓 MCP Client 去調用這個工具，# 然后我們再把工具函數返回的結果給到 DeepSeek，# 讓 DeepSeek 根據工具結果生成最終的回答# 返回最終的回答return '\n'.join(final_text)

上面的代碼中其實針對于沒有使用mcp-server的 prompt功能的tool已經可以了，但是我們在開發 mcp-server demo的時候其實是知道可以配置prompt的，但是實際上我們并沒有根據tool調用的返回去處理 prompt模板。

處理prompt模板

# 嘗試解析工具返回的JSON結果，檢查是否包含MCP模板結構
try:# 將工具返回結果 JSON格式 轉換為 Python 字典tool_result_json = json.loads(tool_result_content)# 檢查是否包含 MCP 模板結構(具有 prompt_template 和 template_args 字段)if(isinstance(tool_result_json, dict) and "prompt_template" in tool_result_json and "template_args" in tool_result_json):raw_data = tool_result_json["raw_data"] # 原始數據prompt_template = tool_result_json["prompt_template"] # 模板函數名稱template_args = tool_result_json["template_args"] # 模板參數# 將模板參數轉換為字符串類型(MCP規范要求)string_args = {k:str(v) for k,v in template_args.items()}# 打印模板參數print(f"模板名稱: {prompt_template}, 模板參數: {string_args}\n--------------------------------\n")# 調用 MCP 服務上的工具，傳入工具名稱和函數參數，返回工具函數執行結果template_response = await self.session.get_prompt(prompt_template, string_args)# 打印工具執行結果，便于調試print(f"模板響應: {template_response}\n--------------------------------\n")if hasattr(template_response, "messages") and template_response.messages:# 打印模板響應print(f"模板具體的信息: {template_response.messages}\n--------------------------------\n")for msg in template_response.messages:# 提取消息內容content = msg.content.text if hasattr(msg.content, "text") else msg.content# 構建歷史信息template_message = {"role": msg.role, # 保持原始角色"content": content # 消息內容}print(f"模板消息歷史: {template_message}\n--------------------------------\n")# 添加到歷史消息中messages.append(template_message)else:print("警告：模板響應中沒有包含消息內容。")
except json.JSONDecodeError:pass# 再次調用 DeepSeek API，讓模型根據工具結果生成最終的回答try:print("正在請求 DeepSeek API 生成最終回答...")# 發送包含工具調用和結果的完整消息歷史final_response = self.llm_client.chat.completions.create(model=self.model, # 指定的模型名稱messages=messages, # 消息歷史（系統提示和用戶問題）temperature=0.5, # 溫度參數，控制響應的隨機性(0.5是中等隨機性)max_tokens=4096 # 最大生成令牌數，限制響應長度)# 添加 DeepSeek 對工具結果的解釋然后到最終輸出final_content = "DeepSeek回答：" + final_response.choices[0].message.contentif final_content:# 如果模型生成了對工具結果的解釋，就將其添加到最終輸出數組中final_text.append(final_content)else:print("警告：DeepSeek API 沒有生成任何內容。")# 如果沒用內容，直接顯示工具結果final_text.append(f"工具調用結果：\n{tool_result_content}")except Exception as e:print(f"生成最終回復時出錯: {e}")final_text.append(f"工具返回結果：\n{tool_result_content}")

啟動client客戶端

async def chat_loop(self):"""運行交互式聊天循環，處理用戶輸入并顯示回復這個函數就是一個簡單的聊天界面，不斷接收用戶輸入，處理問題，并顯示回答，直到用戶輸入'quit'退出。"""print("\nDeepSeek MCP 客戶端已經啟動!")print("請輸入你的問題，輸入'quit'退出。")# 循環處理用戶輸入while True:try:# 獲取用戶輸入query = input("\n問題: ").strip()# 檢查是否要退出if query.lower() == 'quit':break# 處理用戶輸入，傳入到查詢函數中response = await self.process_query(query)print("\n" + response)except Exception as e:print(f"\n錯誤: {str(e)}")async def cleanup(self):"""清理資源，關閉所有打開的連接和上下文。這個函數就像是收拾房間，確保在程序結束時，所有打開的資源都被正常關閉，防止資源泄露。"""# 關閉所有打開的連接和上下文,釋放資源await self.exit_stack.aclose()
async def main():"""主函數，處理命令行參數并啟動客戶端這個函數是程序的起點，它解析命令行參數，創建客戶端實例，連接服務器，并啟動一個聊天循環"""# 檢查命令行參數if len(sys.argv) < 2:print("用法: python client.py <服務器腳本路徑>")sys.exit(1) # 如果參數不足，顯示使用說明并退出# 創建客戶端實例client = DeepSeekMCPClient()try:# 連接到MCP服務器await client.connect_to_server(sys.argv[1])# 啟動聊天循環await client.chat_loop()finally:# 清理資源,確保在任何情況下都清理資源await client.cleanup()# 程序入口點
if __name__ == "__main__":# 運行主函數asyncio.run(main())# 使用說明
# 激活虛擬環境（如果尚未激活）
# source .venv/Scripts/activate# 運行 MCP 客戶端，連接到天氣查詢 MCP 服務器（示例）
# uv run client.py D:\\開源MCP項目\\weather\\weather.py