织梦网站手机版怎么做,网站建设与维护作业,如何用dw制作网页框架,旅行网站开发意义Qwen3接入MCP智能体开发实战#xff08;上#xff09;
一、MCP技术与Qwen3原生MCP能力介绍
1.智能体开发核心技术—MCP
1.1 Function calling技术回顾 如何快速开发一款智能体应用#xff0c;最关键的技术难点就在于如何让大模型高效稳定的接入一些外部工具。而在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 Smitheryhttps://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工具下载的方式最早由ClaudeMCP技术的提出者提出并被广泛接纳。 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_thinkingfalse强制加上 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/ 不写一行代码零门槛接入MCPhttps://www.bilibili.com/video/BV1dCo7YdEgK/ 7分钟开发发布专属MCP【涵盖SSE流式HTTP MCP服务器搭建流程】https://www.bilibili.com/video/BV1VHL6zsE5F/
二、从零到一搭建Qwen3 MCP客户端接入MCP工具 为了更好的为大家展示Qwen3MCP底层原理这里我们先尝试手动搭建一个Qwen3客户端并接入本地或在线的MCP工具。需要注意的是后续我们无论使用哪种Agent开发框架搭建Qwen3MCP的智能体本质上都是这个手动实现流程的更高层的封装与更便捷的实现形式。
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_URLhttp://localhost:11434/v1/
MODELqwen3:30b-a3b-fp16
LLM_API_KEYollama 然后在主函数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(levellogging.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(commandcommand,argsself.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(fError initializing server {self.name}: {e})
await self.cleanup()
raiseasyncdeflist_tools(self) - List[Any]:
获取服务器可用的工具列表Returns:工具列表
ifnot self.session:
raise RuntimeError(fServer {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(fServer {self.name} not initialized)attempt 0
while attempt retries:
try:logging.info(fExecuting {tool_name} on server {self.name}...)result await self.session.call_tool(tool_name, arguments)
return result
except Exception as e:attempt 1logging.warning(
fError executing tool: {e}. Attempt {attempt} of {retries}.)
if attempt retries:logging.info(fRetrying 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(fError 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 []
ifpropertiesin 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_keyapi_key, base_urlbase_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(fError 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) ortypenotin item orfunctionnotin item:
continueold_func item[function]
ifnot isinstance(old_func, dict) ornamenotin old_func ordescriptionnotin old_func:
continuenew_func {
name: old_func[name],
description: old_func[description],
parameters: {}}
ifinput_schemain 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, toolsself.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, toolsself.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, toolsself.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, toolsself.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 导致的异常可以选择忽略
ifAttempted to exit cancel scopein str(e):logging.info(退出时检测到 cancel scope 异常已忽略。)
else:
raiseif __name__ __main__:asyncio.run(main()) 以上客户端为我们团队独家研发的MCP客户端脚本具备以下功能 根据.env配置自由切换底层模型如可接入ollama、vLLM驱动的模型也可以接入OpenAI、DeepSeek等在线模型 根据servers_config.json读取多个MCP服务可以读取本地MCP工具也可以下载在线MCP工具并进行使用 能够进行基于命令行的实现多轮对话 能够同时连接并调用多个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.开启Qwen3MCP调用流程 然后就可以在当前项目的主目录下输入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.0asyncdeffetch_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, paramsparams, headersheaders, timeout30.0)response.raise_for_status()
return response.json() # 返回字典类型
except httpx.HTTPStatusError as e:
return {error: fHTTP 错误: {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}# 如果数据中包含错误信息直接返回错误提示
iferrorin 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(transportstdio) 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.0mcp.tool()
asyncdefwrite_file(content: str) - str:
将指定内容写入本地文件。:param content: 必要参数字符串类型用于表示需要写入文档的具体内容。:return是否成功写入
return已成功写入本地文件。if __name__ __main__:
# 以标准 I/O 方式运行 MCP 服务器mcp.run(transportstdio) 然后需改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工具的流程。 完整项目脚本已上传至课件网盘
