【源碼級開發】Qwen3接入MCP，企業級智能體開發實戰！

Qwen3接入MCP智能體開發實戰（上）

一、MCP技術與Qwen3原生MCP能力介紹

1.智能體開發核心技術—MCP

1.1?Function calling技術回顧

? ? ? ? 如何快速開發一款智能體應用，最關鍵的技術難點就在于如何讓大模型高效穩定的接入一些外部工具。而在MCP技術誕生之前，最主流的方法，是借助Function calling技術來打通大模型和外部工具之間的聯系，換而言之，也就是借助Function calling，來讓大模型靈活的調用外部工具。

? ? ? ? 例如一個典型的Function calling示例，我們希望讓大模型能夠調用一些天氣API查詢即時天氣，此時我們就需要創建一個查詢天氣的外部函數，負責調用天氣API來查詢天氣，同時將外部函數的說明傳遞給大模型，使其能夠根據用戶意圖，在必要的時候向外部函數發起調用請求。

圖片

圖片

? ? ? ? 毫無疑問，Function calling的誕生意義重大，這項技術目前也成為大模型調用外部工具的基本技術范式，哪怕是MCP盛行的今天，底層仍然是Function calling執行流程。

1.2?Qwen3 Function calling能力介紹

? ? ? ? 不過為了更好的學習本期公開課，需要重點強調的是關于大模型的Function calling的能力如何而來。我們都知道，對于當前大模型來說，有些模型有Function calling能力，如DeepSeek-V3模型，而有些模型沒有，如DeepSeek-R1模型：

圖片

而對于Qwen3全系列模型來說，不僅支持Function calling，還支持工具的并聯、串聯調用甚至是自動debug：

圖片

這里我們在公開課參考資料中，為大家整理了一份從零手動實現Qwen3 Function calling的代碼實戰流程：

圖片

圖片

下圖掃碼即可領取：

圖片

本期公開課以Qwen3模型接入MCP技術為主，模型底層function calling實現方法可參考上述資料進行學習。

1.3?Qwen3 Function calling能力從何而來

? ? ? ? 那模型是如何具備Function calling能力的呢？答案是通過模型訓練。對于Qwen3模型來說，由于在訓練階段（指令微調階段）就帶入了大量的類似如下的工具調用對話數據進行訓練，因此能夠識別外部工具并發起對外部工具調用的請求。而類似的，R1模型的訓練過程沒有工具調用數據，因此就不具備Function calling能力。

圖片

而Function calling的能力，是大模型順利開啟MCP功能的基礎。

1.4 MCP技術本質：Function calling的更高層實現

? ? ? ? 而近一段時間大火的MCP技術，其實就可以將其理解為Function calling技術的更高層封裝和實現。傳統的Function calling技術要求圍繞不同的外部工具API單獨創建一個外部函數，類似一把鎖單獨配一把鑰匙，而一個智能體又往往涉及到多個外部工具設計，因此開發工作量很大。

圖片

而MCP技術，全稱為Model Context Protocol，模型上下文協議，是一種開發者共同遵守的協議，在這個協議框架下，大家圍繞某個API開發的外部工具就能夠共用，從而大幅減少重復造輪子的時間。

圖片

2 MCP技術概念介紹

2.1 MCP服務器（server）與客戶端（client）概念介紹

? ? ? ? 不同于Function calling技術，MCP技術是對于大模型和外部工具的另一種劃分方式，也就是說在MCP技術體系中，此時MCP會將外部工具運行腳本稱作服務器，而接入這些外部工具的大模型運行環境稱作客戶端。

圖片

一個客戶端可以接入多個不同類型的服務器的，但要求是都可以遵循MCP通信協議。簡單理解就是MCP服務器的輸出內容是一種標準格式的內容，只能被MCP客戶端所識別。在客戶端和服務器都遵循MCP協議的時候，客戶端就能夠像Function calling中大模型調用外部工具一樣，調用MCP服務器里面的工具。

圖片

2.2 MCP服務器集合

? ? ? ? 暫時拋開底層原理不談，在MCP技術爆發的這幾個月，市面上已經誕生了成百上千的MCP服務器，甚至還出現了大量的MCP服務器集合網站：

  • MCP官方服務器合集：https://github.com/modelcontextprotocol/servers

圖片

  • MCP Github熱門導航：https://github.com/punkpeye/awesome-mcp-servers

圖片

  • Smithery：https://smithery.ai/

圖片

  • MCP導航：https://mcp.so/

圖片

在實際進行智能體開發過程中，我們可以參考這些網站上的MCP工具，并有選擇的對其進行調用。但需要注意的是，無論這些網站的組織形式多么花樣百出，但實際上當我們本地調用MCP工具的時候，都是通過uvx或者npx將對應的庫下載到本地，然后再進行運行。

3. MCP服務器接入示例與標準流程講解

3.1 MCP服務器接入示例

? ? ? ? 而在MCP技術大爆發的今天，接入一個MCP工具也是非常簡單，以下是一個將高德地圖導航MCP（服務器）接入Cherry Studio（客戶端）的示例：

我們能看到，現在如果想要接入一個MCP工具，我們只需要在支持MCP功能的客戶端中寫入相關配置即可。例如我們只需要在Cherry Studio的MCP配置文檔中寫入如下字段：

"amap-maps": {
"isActive":?true,
"command":?"npx",
"args": [
"-y",
"@amap/amap-maps-mcp-server"],
"env": {
"AMAP_MAPS_API_KEY":?"YOUR_API_KRY"},
"name":?"amap-maps"}

圖片

即可讓大模型自動關聯高德MCP工具（服務器），而一個高德MCP服務器的API有幾十種之多：

圖片

可以說是覆蓋了出行生活的放方面。而當一個大模型接入高德MCP服務器后，就能瞬間化身出行規劃智能體。

3.2 MCP工具標準接入流程

? ? ? ? 在上述示例中，我們不難發現，一個MCP服務器標準接入流程是通過寫入一個配置文件來完成的。而在支持MCP功能的客戶端（如Cherry Studio）中寫入MCP工具的配置，其本質就是先將指定的MCP工具下載到本地，然后在有需要的時候對其進行調用。例如高德MCP配置文件如下：

"amap-maps": {
"isActive":?true,
"command":?"npx",
"args": [
"-y",
"@amap/amap-maps-mcp-server"],
"env": {
"AMAP_MAPS_API_KEY":?"YOUR_API_KRY"},
"name":?"amap-maps"}

代表的含義就是我們需要先使用如下命令：

npx-y?@amap/amap-maps-mcp-server

對這個庫@amap/amap-maps-mcp-server進行下載，然后在本地運行，當有必要的時候調用這個庫里面的函數執行相關功能。

? ? ? ? 而這個@amap/amap-maps-mcp-server庫是一個托管在https://www.npmjs.com/上的庫，

圖片

可以使用npx命令進行下載。搜索庫名即可看到這個庫的完整代碼，https://www.npmjs.com/package/@amap/amap-maps-mcp-server：

圖片

而這種通過配置文件來進行MCP工具下載的方式，最早由Claude（MCP技術的提出者）提出并被廣泛接納。

圖片

圖片

4. Qwen3原生MCP能力介紹

4.1 Qwen3原生MCP能力效果

? ? ? ? 而作為新一代最強開源大模型，Qwen 3不僅擁有非常強悍的推理和對話性能，而且為了更好的應對智能體開發需求，Qwen3模型還是全球首款原生支持MCP功能的大模型，能夠更好的理解MCP工具能力、更好的規劃多工具調用流程，因此哪怕面對復雜任務，也能做到游刃有余。

? ? ? ? 不可否認，智能體開發就是當下大模型技術最核心的應用需求，而Qwen3的原生MCP功能，自然就是當下無數技術人最關注的模型功能特性。為了更好的展示模型調用MCP工具的實戰效果，千問官方在模型發布公告中特地展示一段Qwen3-32B模型調用多項MCP工具的實操流程，在這個demo中，用戶要求繪制Qwen模型GitHub項目主頁歷史星標增長曲線圖。

而實際整個任務執行過程非常驚艷，在沒有任何額外的提示的情況下，Qwen3-32B模型調用了包括fetch、time、code-interpreter、filesystem在內的多項MCP工具，并通過這些工具組合調用，完成不同時間點的網站信息檢索、信息采集與本地記錄、借助Python繪制圖像等一系列工作，總耗時約1分鐘左右。而要做到這點，我們僅需讓Qwen3模型接入MCP工具即可。當然為了驗證Qwen3到底有沒有這么強，我們團隊也使用Qwen-chat進行了復現，結果和官方展示的無異。

確實能夠看出Qwen3模型對于外部工具識別和調用能力非常強悍。

4.2 Qwen3原生MCP能力本質與內置提示詞模板解讀

? ? ? ? 其實無論什么模型，所謂MCP能力，指的就是外部工具識別和調用能力，也就是Function calling能力。換而言之，Qwen3的MCP能力強悍，指的就是對于外部工具識別和調用的能力很強。這里我們可以通過觀察Qwen3模型的內置提示詞模板，看出模型是如何識別外部工具的，以下是Qwen3內置提示詞模板解析：

圖片

圖片

Part 1.工具調用（Function Calling）支持部分

{%-?if?tools %}{{-?'<|im_start|>system\n'?}}{%-?ifmessages[0].role ==?'system'?%}{{-?messages[0].content +?'\n\n'?}}{%-?endif?%}{{-?"# Tools\n\nYou may call one or more functions to assist with the user query.\n..."?}}...

解釋：

  • 如果傳入了?tools（即 function calling 的函數簽名），會優先構造?<|im_start|>system?開頭的一段系統提示，告訴模型可以調用工具。

  • 這段提示包含：

    • # Tools?開頭的說明文字；

    • tools?列表，每個工具（函數）都通過?tojson?轉換為 JSON；

    • 如何使用?<tool_call>?標簽返回工具調用的結果。

Part 2.系統消息處理

{%-?ifmessages[0].role ==?'system'?%}{{-?'<|im_start|>system\n'?+?messages[0].content +?'<|im_end|>\n'?}}
{%-?endif?%}

解釋：

  • 如果首條消息是?system，則會作為系統設定（system prompt）處理，加上?<|im_start|>system\n ... <|im_end|>\n。

Part 3.多輪消息回顯處理

{%-?for?message in?messages?%}{%-?if?(message.role ==?"user") ... %}{{-?'<|im_start|>'?+ message.role +?'\n'?+ message.content +?'<|im_end|>'?+?'\n'?}}

解釋：

  • 針對用戶（user）、助手（assistant）、工具響應（tool）等不同角色進行處理。

  • 使用?<|im_start|>role\n...<|im_end|>?包裹每一輪對話。

4Assistant 角色的特殊處理（含推理內容）

{%-?if?message.role ==?"assistant"?%}...
<think>\n...reasoning_content...\n</think>

解釋：

  • 若助手消息中包含?<think>?內容，會將其拆分為“推理部分”和“回復正文”。

  • 如果存在 tool_calls，還會附加一段?<tool_call>?JSON 標簽。

Part 5.工具響應處理（role = tool）

<tool_response>\n...內容...\n</tool_response>

解釋：

  • 模型回復?<tool_call>?后，你會給出?<tool_response>。

  • 這部分內容會包在 user role 內部，以?<tool_response>?標簽封裝，用來模擬用戶獲得工具調用結果。

Part 6.混合推理模式開啟方法

{%-?if?add_generation_prompt %}{{-?'<|im_start|>assistant\n'?}}{%-?if?enable_thinking?is?defined?and?enable_thinking?is?false %}{{-?'<think>\n\n</think>\n\n'?}}{%-?endif?%}
{%-?endif?%}

解釋：

  • 如果需要生成下一輪回復，會在最后加上?<|im_start|>assistant\n?作為提示。

  • 還可以通過設置?enable_thinking=false，強制加上?<think>?占位符。

5. Qwen3模型借助桌面端應用快速接入MCP工具

? ? ? ? 在了解了基本遠離后，接下來我們即可使用桌面端應用Cherry Studio，快速上手嘗試使用Qwen3接入MCP工具，例如先通過ollama下載模型并開啟ollama服務，然后使用Cherry Studio配置天氣查詢MCP工具進行測試，演示效果如下：

6.公開課大綱及課件領取

  • 公開課課件領取

圖片

  • MCP系列公開課參考

在此前我已經開設過一些列公開課來介紹MCP工具，相關基礎內容大家可以參考如下公開課：

圖片

  • MCP技術開發入門：https://www.bilibili.com/video/BV1NLXCYTEbj/

  • MCP企業級智能體開發實戰：https://www.bilibili.com/video/BV1n1ZuYjEzf/

  • 不寫一行代碼，零門檻接入MCP：https://www.bilibili.com/video/BV1dCo7YdEgK/

  • 7分鐘開發&發布專屬MCP！【涵蓋SSE&流式HTTP MCP服務器搭建流程】：https://www.bilibili.com/video/BV1VHL6zsE5F/

二、從零到一搭建Qwen3 MCP客戶端接入MCP工具

? ? ? ? 為了更好的為大家展示Qwen3+MCP底層原理，這里我們先嘗試手動搭建一個Qwen3客戶端，并接入本地或在線的MCP工具。需要注意的是，后續我們無論使用哪種Agent開發框架，搭建Qwen3+MCP的智能體，本質上都是這個手動實現流程的更高層的封裝與更便捷的實現形式。

1. 基礎環境搭建

? ? ? ? 這里我們采用uv工具進行Python項目管理，首先進入到某個自由選定的目錄下，并使用uv進行項目初始化：

#cd?/root/autodl-tmp/Qwen3#?創建項目目錄
uv init Qwen3-MCP
cd Qwen3-MCP

圖片

pip?install?uv

然后輸入如下命令創建虛擬環境：

#?創建虛擬環境
uv venv#?激活虛擬環境
source .venv/bin/activate

此時就構建了一個基礎項目結構：

圖片

最后需要添加如下依賴：

uv?add?httpx openai mcp

圖片

2. 編寫基于Qwen3的MCP客戶端

? ? ? ? 首先設置配置文件，在當前目錄創建.env文件，寫入ollama驅動下的Qwen3模型調用地址和模型名稱：

圖片

圖片

BASE_URL=http://localhost:11434/v1/
MODEL=qwen3:30b-a3b-fp16
LLM_API_KEY=ollama

圖片

然后在主函數main.py內寫入如下內容：

import?asyncio
import?json
import?logging
import?os
import?shutil
from?contextlib?import?AsyncExitStack
from?typing?import?Any, Dict, List, Optionalimport?httpx
from?dotenv?import?load_dotenv
from?openai?import?OpenAI ?# OpenAI Python SDK
from?mcp?import?ClientSession, StdioServerParameters
from?mcp.client.stdio?import?stdio_client# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)# =============================
# 配置加載類（支持環境變量及配置文件）
# =============================
classConfiguration:
"""管理 MCP 客戶端的環境變量和配置文件"""def__init__(self)?->?None:load_dotenv()
# 從環境變量中加載 API key, base_url 和 modelself.api_key = os.getenv("LLM_API_KEY")self.base_url = os.getenv("BASE_URL")self.model = os.getenv("MODEL")
ifnot?self.api_key:
raise?ValueError("? 未找到 LLM_API_KEY，請在 .env 文件中配置")@staticmethod
defload_config(file_path: str)?-> Dict[str, Any]:
"""從 JSON 文件加載服務器配置Args:file_path: JSON 配置文件路徑Returns:包含服務器配置的字典"""
with?open(file_path,?"r")?as?f:
return?json.load(f)# =============================
# MCP 服務器客戶端類
# =============================
classServer:
"""管理單個 MCP 服務器連接和工具調用"""def__init__(self, name: str, config: Dict[str, Any])?->?None:self.name: str = nameself.config: Dict[str, Any] = configself.session: Optional[ClientSession] =?Noneself.exit_stack: AsyncExitStack = AsyncExitStack()self._cleanup_lock = asyncio.Lock()asyncdefinitialize(self)?->?None:
"""初始化與 MCP 服務器的連接"""
# command 字段直接從配置獲取command = self.config["command"]
if?command?isNone:
raise?ValueError("command 不能為空")server_params = StdioServerParameters(command=command,args=self.config["args"],env={**os.environ, **self.config["env"]}?if?self.config.get("env")?elseNone,)
try:stdio_transport =?await?self.exit_stack.enter_async_context(stdio_client(server_params))read_stream, write_stream = stdio_transportsession =?await?self.exit_stack.enter_async_context(ClientSession(read_stream, write_stream))
await?session.initialize()self.session = session
except?Exception?as?e:logging.error(f"Error initializing server?{self.name}:?{e}")
await?self.cleanup()
raiseasyncdeflist_tools(self)?-> List[Any]:
"""獲取服務器可用的工具列表Returns:工具列表"""
ifnot?self.session:
raise?RuntimeError(f"Server?{self.name}?not initialized")tools_response =?await?self.session.list_tools()tools = []
for?item?in?tools_response:
if?isinstance(item, tuple)?and?item[0] ==?"tools":
for?tool?in?item[1]:tools.append(Tool(tool.name, tool.description, tool.inputSchema))
return?toolsasyncdefexecute_tool(self, tool_name: str, arguments: Dict[str, Any], retries: int =?2, delay: float =?1.0)?-> Any:
"""執行指定工具，并支持重試機制Args:tool_name: 工具名稱arguments: 工具參數retries: 重試次數delay: 重試間隔秒數Returns:工具調用結果"""
ifnot?self.session:
raise?RuntimeError(f"Server?{self.name}?not initialized")attempt =?0
while?attempt < retries:
try:logging.info(f"Executing?{tool_name}?on server?{self.name}...")result =?await?self.session.call_tool(tool_name, arguments)
return?result
except?Exception?as?e:attempt +=?1logging.warning(
f"Error executing tool:?{e}. Attempt?{attempt}?of?{retries}.")
if?attempt < retries:logging.info(f"Retrying in?{delay}?seconds...")
await?asyncio.sleep(delay)
else:logging.error("Max retries reached. Failing.")
raiseasyncdefcleanup(self)?->?None:
"""清理服務器資源"""
asyncwith?self._cleanup_lock:
try:
await?self.exit_stack.aclose()self.session =?None
except?Exception?as?e:logging.error(f"Error during cleanup of server?{self.name}:?{e}")# =============================
# 工具封裝類
# =============================
classTool:
"""封裝 MCP 返回的工具信息"""def__init__(self, name: str, description: str, input_schema: Dict[str, Any])?->?None:self.name: str = nameself.description: str = descriptionself.input_schema: Dict[str, Any] = input_schemadefformat_for_llm(self)?-> str:
"""生成用于 LLM 提示的工具描述"""args_desc = []
if"properties"in?self.input_schema:
for?param_name, param_info?in?self.input_schema["properties"].items():arg_desc =?f"-?{param_name}:?{param_info.get('description',?'No description')}"
if?param_name?in?self.input_schema.get("required", []):arg_desc +=?" (required)"args_desc.append(arg_desc)
returnf"""
Tool:?{self.name}
Description:?{self.description}
Arguments:
{chr(10).join(args_desc)}
"""# =============================
# LLM 客戶端封裝類（使用 OpenAI SDK）
# =============================
classLLMClient:
"""使用 OpenAI SDK 與大模型交互"""def__init__(self, api_key: str, base_url: Optional[str], model: str)?->?None:self.client = OpenAI(api_key=api_key, base_url=base_url)self.model = modeldefget_response(self, messages: List[Dict[str, Any]], tools: Optional[List[Dict[str, Any]]] = None)?-> Any:
"""發送消息給大模型 API，支持傳入工具參數（function calling 格式）"""payload = {
"model": self.model,
"messages": messages,
"tools": tools,}
try:response = self.client.chat.completions.create(**payload)
return?response
except?Exception?as?e:logging.error(f"Error during LLM call:?{e}")
raise# =============================
# 多服務器 MCP 客戶端類（集成配置文件、工具格式轉換與 OpenAI SDK 調用）
# =============================
classMultiServerMCPClient:
def__init__(self)?->?None:
"""管理多個 MCP 服務器，并使用 OpenAI Function Calling 風格的接口調用大模型"""self.exit_stack = AsyncExitStack()config = Configuration()self.openai_api_key = config.api_keyself.base_url = config.base_urlself.model = config.modelself.client = LLMClient(self.openai_api_key, self.base_url, self.model)
# (server_name -> Server 對象)self.servers: Dict[str, Server] = {}
# 各個 server 的工具列表self.tools_by_server: Dict[str, List[Any]] = {}self.all_tools: List[Dict[str, Any]] = []asyncdefconnect_to_servers(self, servers_config: Dict[str, Any])?->?None:
"""根據配置文件同時啟動多個服務器并獲取工具servers_config 的格式為：{"mcpServers": {"sqlite": { "command": "uvx", "args": [ ... ] },"puppeteer": { "command": "npx", "args": [ ... ] },...}}"""mcp_servers = servers_config.get("mcpServers", {})
for?server_name, srv_config?in?mcp_servers.items():server = Server(server_name, srv_config)
await?server.initialize()self.servers[server_name] = servertools =?await?server.list_tools()self.tools_by_server[server_name] = toolsfor?tool?in?tools:
# 統一重命名：serverName_toolNamefunction_name =?f"{server_name}_{tool.name}"self.all_tools.append({
"type":?"function",
"function": {
"name": function_name,
"description": tool.description,
"input_schema": tool.input_schema}})# 轉換為 OpenAI Function Calling 所需格式self.all_tools =?await?self.transform_json(self.all_tools)logging.info("\n? 已連接到下列服務器:")
for?name?in?self.servers:srv_cfg = mcp_servers[name]logging.info(f" ?-?{name}: command={srv_cfg['command']}, args={srv_cfg['args']}")logging.info("\n匯總的工具:")
for?t?in?self.all_tools:logging.info(f" ?-?{t['function']['name']}")asyncdeftransform_json(self, json_data: List[Dict[str, Any]])?-> List[Dict[str, Any]]:
"""將工具的 input_schema 轉換為 OpenAI 所需的 parameters 格式，并刪除多余字段"""result = []
for?item?in?json_data:
ifnot?isinstance(item, dict)?or"type"notin?item?or"function"notin?item:
continueold_func = item["function"]
ifnot?isinstance(old_func, dict)?or"name"notin?old_func?or"description"notin?old_func:
continuenew_func = {
"name": old_func["name"],
"description": old_func["description"],
"parameters": {}}
if"input_schema"in?old_func?and?isinstance(old_func["input_schema"], dict):old_schema = old_func["input_schema"]new_func["parameters"]["type"] = old_schema.get("type",?"object")new_func["parameters"]["properties"] = old_schema.get("properties", {})new_func["parameters"]["required"] = old_schema.get("required", [])new_item = {
"type": item["type"],
"function": new_func}result.append(new_item)
return?resultasyncdefchat_base(self, messages: List[Dict[str, Any]])?-> Any:
"""使用 OpenAI 接口進行對話，并支持多次工具調用（Function Calling）。如果返回 finish_reason 為 "tool_calls"，則進行工具調用后再發起請求。"""response = self.client.get_response(messages, tools=self.all_tools)
# 如果模型返回工具調用
if?response.choices[0].finish_reason ==?"tool_calls":
whileTrue:messages =?await?self.create_function_response_messages(messages, response)response = self.client.get_response(messages, tools=self.all_tools)
if?response.choices[0].finish_reason !=?"tool_calls":
break
return?responseasyncdefcreate_function_response_messages(self, messages: List[Dict[str, Any]], response: Any)?-> List[Dict[str, Any]]:
"""將模型返回的工具調用解析執行，并將結果追加到消息隊列中"""function_call_messages = response.choices[0].message.tool_callsmessages.append(response.choices[0].message.model_dump())
for?function_call_message?in?function_call_messages:tool_name = function_call_message.function.nametool_args = json.loads(function_call_message.function.arguments)
# 調用 MCP 工具function_response =?await?self._call_mcp_tool(tool_name, tool_args)messages.append({
"role":?"tool",
"content": function_response,
"tool_call_id": function_call_message.id,})
return?messagesasyncdefprocess_query(self, user_query: str)?-> str:
"""OpenAI Function Calling 流程：1. 發送用戶消息 + 工具信息2. 若模型返回 finish_reason 為 "tool_calls"，則解析并調用 MCP 工具3. 將工具調用結果返回給模型，獲得最終回答"""messages = [{"role":?"user",?"content": user_query}]response = self.client.get_response(messages, tools=self.all_tools)content = response.choices[0]logging.info(content)
if?content.finish_reason ==?"tool_calls":tool_call = content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)logging.info(f"\n[ 調用工具:?{tool_name}, 參數:?{tool_args}?]\n")result =?await?self._call_mcp_tool(tool_name, tool_args)messages.append(content.message.model_dump())messages.append({
"role":?"tool",
"content": result,
"tool_call_id": tool_call.id,})response = self.client.get_response(messages, tools=self.all_tools)
return?response.choices[0].message.content
return?content.message.contentasyncdef_call_mcp_tool(self, tool_full_name: str, tool_args: Dict[str, Any])?-> str:
"""根據 "serverName_toolName" 格式調用相應 MCP 工具"""parts = tool_full_name.split("_",?1)
if?len(parts) !=?2:
returnf"無效的工具名稱:?{tool_full_name}"server_name, tool_name = partsserver = self.servers.get(server_name)
ifnot?server:
returnf"找不到服務器:?{server_name}"resp =?await?server.execute_tool(tool_name, tool_args)
return?resp.content?if?resp.content?else"工具執行無輸出"asyncdefchat_loop(self)?->?None:
"""多服務器 MCP + OpenAI Function Calling 客戶端主循環"""logging.info("\n🤖 多服務器 MCP + Function Calling 客戶端已啟動！輸入 'quit' 退出。")messages: List[Dict[str, Any]] = []
whileTrue:query = input("\n你: ").strip()
if?query.lower() ==?"quit":
break
try:messages.append({"role":?"user",?"content": query})messages = messages[-20:] ?# 保持最新 20 條上下文response =?await?self.chat_base(messages)messages.append(response.choices[0].message.model_dump())result = response.choices[0].message.content
# logging.info(f"\nAI: {result}")print(f"\nAI:?{result}")
except?Exception?as?e:print(f"\n?? ?調用過程出錯:?{e}")asyncdefcleanup(self)?->?None:
"""關閉所有資源"""
await?self.exit_stack.aclose()# =============================
# 主函數
# =============================
asyncdefmain()?->?None:
# 從配置文件加載服務器配置config = Configuration()servers_config = config.load_config("servers_config.json")client = MultiServerMCPClient()
try:
await?client.connect_to_servers(servers_config)
await?client.chat_loop()
finally:
try:
await?asyncio.sleep(0.1)
await?client.cleanup()
except?RuntimeError?as?e:
# 如果是因為退出 cancel scope 導致的異常，可以選擇忽略
if"Attempted to exit cancel scope"in?str(e):logging.info("退出時檢測到 cancel scope 異常，已忽略。")
else:
raiseif?__name__ ==?"__main__":asyncio.run(main())

圖片

以上客戶端為我們團隊獨家研發的MCP客戶端腳本，具備以下功能：

  1. 根據.env配置，自由切換底層模型，如可接入ollama、vLLM驅動的模型，也可以接入OpenAI、DeepSeek等在線模型；

  2. 根據servers_config.json讀取多個MCP服務，可以讀取本地MCP工具，也可以下載在線MCP工具并進行使用；

  3. 能夠進行基于命令行的實現多輪對話；

  4. 能夠同時連接并調用多個MCP server上的多個工具，并能實現多工具的并聯和串聯調用；

3. 配置MCP服務器

? ? ? ? 接下來繼續創建一個servers_config.json腳本，并寫入我們希望調用的MCP工具，例如我們想要調用Filesystem MCP，該MCP是一個最基礎同時也是最常用的MCP服務器，同時也是官方推薦的服務器，服務器項目地址：https://github.com/modelcontextprotocol/servers/tree/main/src/filesystem

圖片

借助Filesystem，我們可以高效便捷操作本地文件夾。同時Filesystem也是一個js項目，源碼托管在npm平臺上：（https://www.npmjs.com/package/@modelcontextprotocol/server-filesystem）

圖片

我們直接創建servers_config.josn并寫入如下配置即可調用：

{
"mcpServers": {
"filesystem": {
"command":?"npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/path/to/other/allowed/dir"]}}
}

圖片

圖片

4.開啟Qwen3+MCP調用流程

? ? ? ? 然后就可以在當前項目的主目錄下輸入uv run進行運行：

uv run main.py

圖片

能夠看到，此時已經關聯了filesystem多項外部工具。接下來進行對話：

圖片

由于模型開啟了思考模式，所以能看到的具體過程。

? ? ? ? 接下來繼續嘗試調用外部工具：

圖片

能夠看到，Qwen3 MCP Client能夠順利調用外部工具。

? ? ? ? 此外，我們還可以讓Qwen3 MCP Client接入自定義的函數，例如我們在當前項目中創建兩個MCP server腳本：

  • weather_server.py：查詢天氣MCP服務器

圖片

并寫入如下內容

import?os
import?json
import?httpx
from?typing?import?Any
from?dotenv?import?load_dotenv
from?mcp.server.fastmcp?import?FastMCP# 初始化 MCP 服務器
mcp = FastMCP("WeatherServer")# OpenWeather API 配置
OPENWEATHER_API_BASE =?"https://api.openweathermap.org/data/2.5/weather"
API_KEY =?"YOUR_API_KEY"# 填寫你的OpenWeather-API-KEY
USER_AGENT =?"weather-app/1.0"asyncdeffetch_weather(city: str)?-> dict[str, Any] |?None:
"""從 OpenWeather API 獲取天氣信息。:param city: 城市名稱（需使用英文，如 Beijing）:return: 天氣數據字典；若出錯返回包含 error 信息的字典"""params = {
"q": city,
"appid": API_KEY,
"units":?"metric",
"lang":?"zh_cn"}headers = {"User-Agent": USER_AGENT}asyncwith?httpx.AsyncClient()?as?client:
try:response =?await?client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0)response.raise_for_status()
return?response.json() ?# 返回字典類型
except?httpx.HTTPStatusError?as?e:
return?{"error":?f"HTTP 錯誤:?{e.response.status_code}"}
except?Exception?as?e:
return?{"error":?f"請求失敗:?{str(e)}"}defformat_weather(data: dict[str, Any] | str)?-> str:
"""將天氣數據格式化為易讀文本。:param data: 天氣數據（可以是字典或 JSON 字符串）:return: 格式化后的天氣信息字符串"""
# 如果傳入的是字符串，則先轉換為字典
if?isinstance(data, str):
try:data = json.loads(data)
except?Exception?as?e:
returnf"無法解析天氣數據:?{e}"# 如果數據中包含錯誤信息，直接返回錯誤提示
if"error"in?data:
returnf"???{data['error']}"# 提取數據時做容錯處理city = data.get("name",?"未知")country = data.get("sys", {}).get("country",?"未知")temp = data.get("main", {}).get("temp",?"N/A")humidity = data.get("main", {}).get("humidity",?"N/A")wind_speed = data.get("wind", {}).get("speed",?"N/A")
# weather 可能為空列表，因此用 [0] 前先提供默認字典weather_list = data.get("weather", [{}])description = weather_list[0].get("description",?"未知")return?(
f"🌍?{city},?{country}\n"
f"🌡 溫度:?{temp}°C\n"
f"💧 濕度:?{humidity}%\n"
f"🌬 風速:?{wind_speed}?m/s\n"
f"🌤 天氣:?{description}\n")@mcp.tool()
asyncdefquery_weather(city: str)?-> str:
"""輸入指定城市的英文名稱，返回今日天氣查詢結果。:param city: 城市名稱（需使用英文）:return: 格式化后的天氣信息"""data =?await?fetch_weather(city)
return?format_weather(data)if?__name__ ==?"__main__":
# 以標準 I/O 方式運行 MCP 服務器mcp.run(transport='stdio')

圖片

  • write_server.py：寫入本地文檔MCP服務器

圖片

并寫入如下內容

import?json
import?httpx
from?typing?import?Any
from?mcp.server.fastmcp?import?FastMCP# 初始化 MCP 服務器
mcp = FastMCP("WriteServer")
USER_AGENT =?"write-app/1.0"@mcp.tool()
asyncdefwrite_file(content: str)?-> str:
"""將指定內容寫入本地文件。:param content: 必要參數，字符串類型，用于表示需要寫入文檔的具體內容。:return：是否成功寫入"""
return"已成功寫入本地文件。"if?__name__ ==?"__main__":
# 以標準 I/O 方式運行 MCP 服務器mcp.run(transport='stdio')

圖片

然后需改servers_config.json配置文件，寫入如下內容：

{
"mcpServers": {
"filesystem": {
"command":?"npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"/root/autodl-tmp/Qwen3/Qwen3-MCP"]},
"weather": {
"command":?"python",
"args": ["weather_server.py"]},
"write": {
"command":?"python",
"args": ["write_server.py"]}}
}

圖片

然后再次開啟對話，就能看到加載了更多工具進來：

圖片

并可進行多MCP服務器的多工具并行調用：

圖片

圖片

和串聯調用：

圖片

圖片

至此我們就完成了Qwen-3接入在線MCP工具的流程。

圖片

  • 完整項目腳本已上傳至課件網盤

圖片

本文來自互聯網用戶投稿，該文觀點僅代表作者本人，不代表本站立場。本站僅提供信息存儲空間服務，不擁有所有權，不承擔相關法律責任。
如若轉載，請注明出處：http://www.pswp.cn/news/905756.shtml
繁體地址，請注明出處：http://hk.pswp.cn/news/905756.shtml
英文地址，請注明出處：http://en.pswp.cn/news/905756.shtml

如若內容造成侵權/違法違規/事實不符，請聯系多彩編程網進行投訴反饋email:809451989@qq.com，一經查實，立即刪除！

相關文章

Linux下載與安裝

Linux下載與安裝

一、YUM 1.1 什么是YUM 在CentOS系統中&#xff0c;軟件管理方式通常有三種方式&#xff1a;rpm安裝、yum安裝以及編譯&#xff08;源碼&#xff09;安裝。 編譯安裝&#xff0c;從過程上來講比較麻煩&#xff0c;包需要用戶自行下載&#xff0c;下載的是源碼包&#xff0c;需…
閱讀更多...
PostgreSQL中的全頁寫

PostgreSQL中的全頁寫

一、概述 在PGSQL數據庫中&#xff0c;默認的頁面大小為8KB&#xff0c;但是磁盤buffer的大小為4KB&#xff0c;扇區大小為512B。這就導致在操作系統的角度看數據庫的寫操作&#xff0c;其實并不是一種原子操作。如果操作系統發生了系統級別的故障&#xff0c;此時正好操作系統…
閱讀更多...
WEB安全--Java安全--shiro550反序列化漏洞

WEB安全--Java安全--shiro550反序列化漏洞

一、前言 什么是shiro&#xff1f; shiro是一個Apache的Java安全框架 它的作用是什么&#xff1f; Apache Shiro 是一個強大且靈活的 Java 安全框架&#xff0c;用于處理身份驗證、授權、密碼管理以及會話管理等功能 二、shiro550反序列化原理 1、用戶首次登錄并勾選記住密碼…
閱讀更多...
2024 睿抗機器人開發者大賽CAIP-編程技能賽-專科組（國賽）解題報告 | 珂學家

2024 睿抗機器人開發者大賽CAIP-編程技能賽-專科組（國賽）解題報告 | 珂學家

前言 題解 2024 睿抗機器人開發者大賽CAIP-編程技能賽-專科組&#xff08;國賽&#xff09;&#xff0c;陳越姐姐出題。 國賽比省賽&#xff0c;難度增強了不少&#xff0c;題目就剩下4個題了。 涉及堆棧&#xff0c;hash表&#xff0c;優先隊列等高階數據結構的使用&#x…
閱讀更多...
15 C 語言字符類型詳解：轉義字符、格式化輸出、字符類型本質、ASCII 碼編程實戰、最值宏匯總

15 C 語言字符類型詳解：轉義字符、格式化輸出、字符類型本質、ASCII 碼編程實戰、最值宏匯總

1 字符類型概述 在 C 語言中&#xff0c;字符類型 char 用于表示單個字符&#xff0c;例如一個數字、一個字母或一個符號。 char 類型的字面量是用單引號括起來的單個字符&#xff0c;例如 A、5 或 #。 當需要表示多個字符組成的序列時&#xff0c;就涉及到了字符串。在 C 語言…
閱讀更多...
操作系統-鎖/內存/中斷/IO

操作系統-鎖/內存/中斷/IO

文章目錄 鎖自旋鎖互斥鎖悲觀鎖和樂觀鎖 內存管理物理/虛擬內存頁表段表虛擬內存布局寫時復制copy on writebrk&#xff0c;mmap頁面置換算法 中斷中斷分類中斷流程 網絡I/OI/O模型服務器處理并發請求 鎖 自旋鎖 自旋鎖是一種基于忙等待&#xff08;Busy-Waiting&#xff09;…
閱讀更多...
割點與其例題

割點與其例題

割點 定義&#xff1a; 若一個點在圖中被去掉后&#xff0c;圖的連通塊個數增加&#xff0c;那么這個點就被稱為“割點”。如下圖所示紅點。 定義說白了就是若去掉一個點&#xff0c;圖被“斷開”的點稱為割點。 樸素算法&#xff1a; 枚舉每個點 u。遍歷圖&#xff0c;如果…
閱讀更多...
圖卷積神經網絡（Graph Convolutional Network, GCN）

圖卷積神經網絡（Graph Convolutional Network, GCN）

最近看論文看到了圖卷積神經網絡的內容&#xff0c;之前整理過圖神經網絡的內容&#xff0c;這里再補充一下&#xff0c;方便以后查閱。 圖卷積神經網絡&#xff08;Graph Convolutional Network, GCN&#xff09; 圖卷積神經網絡1. 什么是圖卷積神經網絡&#xff08;GCN&#…
閱讀更多...
安裝win11硬盤分區MBR還是GPT_裝win11系統分區及安裝教程

安裝win11硬盤分區MBR還是GPT_裝win11系統分區及安裝教程

最近有網友問我,裝win11系統分區有什么要求裝win11系統硬盤分區用mbr還是GPT&#xff1f;我們知道現在的引導模式有uefi和legacy兩種引導模式&#xff0c;如果采用的是uefi引導模式&#xff0c;分區類型對應的就是gpt分區(guid)&#xff0c;如果引導模式采用的是legacy&#xf…
閱讀更多...
服務培訓QDA 的安裝調試方法，硬件模塊的講解和軟件控制臺使用及系統測試

服務培訓QDA 的安裝調試方法，硬件模塊的講解和軟件控制臺使用及系統測試

#服務培訓##質譜儀##軟件控制##硬件模塊# 以下是關于Waters QDa單桿液質質譜儀的安裝調試、硬件模塊講解以及軟件控制臺使用培訓的相關內容&#xff1a; 安裝調試 場地準備&#xff1a;用戶需要提前準備好實驗室&#xff0c;確保實驗室環境符合儀器的要求&#xff0c;如溫度、…
閱讀更多...
在K8S集群中部署EFK日志收集

在K8S集群中部署EFK日志收集

目錄 引言環境準備安裝自定義資源部署ElasticsearchMaster 節點與 Data 節點的區別生產優化建議安裝好以后測試ES是否正常部署Fluentd測試filebeat是否正常推送日志部署Kibana獲取賬號密碼&#xff0c;賬號是&#xff1a;elastic集群測試 引言 系統版本為 Centos7.9內核版本為…
閱讀更多...
polarctf-web-[rce1]

polarctf-web-[rce1]

考點&#xff1a; (1)RCE(exec函數) (2)空格繞過 (3)執行函數(exec函數) (4)閉合(ping命令閉合) 題目來源&#xff1a;Polarctf-web-[rce1] 解題&#xff1a; 這段代碼實現了一個簡單的 Ping 測試工具&#xff0c;用戶可以通過表單提交一個 IP 地址&#xff0c;服務器會執…
閱讀更多...
【串流VR手勢】Pico 4 Ultra Enterprise 在 SteamVR 企業串流中無法識別手勢的問題排查與解決過程（Pico4UE串流手勢問題）

【串流VR手勢】Pico 4 Ultra Enterprise 在 SteamVR 企業串流中無法識別手勢的問題排查與解決過程（Pico4UE串流手勢問題）

寫在前面的話 此前&#xff08;用Pico 4U&#xff09;接入了MRTK3&#xff0c;現項目落地需要部署&#xff0c;發現串流場景中&#xff0c;Pico4UE的企業串流無法正常識別手勢。&#xff08;一體機方式部署使用無問題&#xff09; 花了半小時解決&#xff0c;怕忘&#xff0c;…
閱讀更多...
ES（Elasticsearch）的應用與代碼示例

ES（Elasticsearch）的應用與代碼示例

Elasticsearch應用與代碼示例技術文章大綱 一、引言 Elasticsearch在現代化應用中的核心作用典型應用場景分析&#xff08;日志分析/全文檢索/數據聚合&#xff09; 二、環境準備(前提條件) Elasticsearch 8.x集群部署要點IK中文分詞插件配置指南Ingest Attachment插件安裝…
閱讀更多...
臨床決策支持系統的提示工程優化路徑深度解析

臨床決策支持系統的提示工程優化路徑深度解析

引言 隨著人工智能技術在醫療領域的迅猛發展，臨床決策支持系統(CDSS)正經歷從傳統規則引擎向智能提示工程的范式轉變。在這一背景下，如何構建既符合循證醫學原則又能適應個體化醫療需求的CDSS成為醫學人工智能領域的核心挑戰。本報告深入剖析了臨床決策支持系統中提示工程的…
閱讀更多...
火山RTC 8 SDK集成進項目中

火山RTC 8 SDK集成進項目中

一、SDK 集成預備工作 1、SDK下載 https://www.volcengine.com/docs/6348/75707 2、解壓后 3、放在自己項目中的位置 1&#xff09;、include 2&#xff09;、lib 3)、dll 暫時&#xff0c;只需要VolcEngineRTC.dll RTCFFmpeg.dll openh264-4.dll&#xff0c; 放在intLive2…
閱讀更多...
OkHttp用法-Java調用http服務

OkHttp用法-Java調用http服務

特點&#xff1a;高性能&#xff0c;支持異步請求&#xff0c;連接池優化 官方文檔&#xff1a;提供快速入門指南和高級功能&#xff08;如攔截器、連接池&#xff09;的詳細說明&#xff0c;GitHub倉庫包含豐富示例。 社區資源&#xff1a;中文教程豐富&#xff0c;GitHub高…
閱讀更多...
python中常用的參數以及命名規范

python中常用的參數以及命名規范

以下是 Python 中常見的命名規范、參數用法及在大型項目中常用的操作模式&#xff0c;供記錄參考&#xff1a; 1. 命名規范&#xff08;Naming Conventions&#xff09; 前綴/形式含義示例_age單下劃線&#xff1a;弱“私有”標記&#xff08;可訪問但不建議外部使用&#xff…
閱讀更多...
第五十七篇 Java接口設計之道：從咖啡機到智能家居的編程哲學

第五十七篇 Java接口設計之道：從咖啡機到智能家居的編程哲學

目錄 引言&#xff1a;生活中的接口無處不在一、咖啡機與基礎接口&#xff1a;理解抽象契約1.1 咖啡制作的標準接口 二、智能家居與策略模式&#xff1a;靈活切換實現2.1 溫度調節策略場景 三、物流系統與工廠模式&#xff1a;標準接口下的多樣實現3.1 快遞運輸接口設計 四、健…
閱讀更多...
第二十六天打卡

第二十六天打卡

全局變量 global_var 全局變量是定義在函數、類或者代碼塊外部的變量&#xff0c;它在整個程序文件內都能被訪問。在代碼里&#xff0c; global_var 就是一個全局變量&#xff0c;下面是相關代碼片段&#xff1a; print("\n--- 變量作用域示例 ---") global_var …
閱讀更多...
最新文章

Copyright @ 2022~2023