mcp 就是一个接口，能为大模型引入各种丰富的功能，目前公开的 mcp 服务很多，能实现很多复杂的操作。之前曾用好几篇帖子来记录mcp的使用，也尝试使用了一些开源的mcp server，除了新鲜感作祟，更多的是对于将AI用于具体工作的思考。可用了很多别人的工具，我也就开始考虑开发专属于自己的mcp服务，毕竟有的工作具有极强的针对性，不是公共工具可以轻松完成的。仔细研究下来，开发自己的mcp工具其实并不复杂。个人愚见，能用Python等脚本实现的功能，只要有一个输入和输出，似乎就可以利用mcp库与mcp client建立联系，从而为AI所用。于是我就以 dify 工作流为切入口，开发一个可以用 mcp 直接操纵 dify 工作流的工具。
以下服务依赖一些先导条件：
- 我采用本地部署的 dify 完成工作流（直接使用 dify 官方云服务也可以，接口改一下就行）。
- 本地配置 uv 、node.js 环境。
我们最重要的是需要这个库：
from mcp.server.fastmcp import FastMCP其他库就根据你自己的脚本需求安装就好了。
搭建 dify 工作流
搭建工作流不是本教程的核心，我展示一下我搭建的一个基础工作流：
功能非常简单，就是用户输入，然后调用本地部署的 searxng 进行查询，然后再用大模型整合后输出。
在“访问 API”页面中可以查询 api 使用文档：
我们就是用本地的http://127.0.0.1/v1/chat-messages的端口进行信息获取。
搭建脚本环境
参考链接：
https://modelcontextprotocol.io/quickstart/server#weather-api-issues
安装 uv，在 power shell 中执行：
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"然后配置 uv 环境路径：
- Win + R，输入
sysdm.cpl，按 Enter。 - 进入 “高级” → “环境变量”。
- 在 “用户变量” 部分找到
Path，双击它。 - 点击“新建”，添加
C:\Users\{你的用户名}\.local\bin。 - 确认后关闭窗口，并重新打开 PowerShell 终端。
重现打开 power shell，执行如下命令确认安装完成：
uv --version
创建 mcp 项目
在一个你想创建项目的文件夹，右键打开 power shell 执行如下命令：
uv init mcptool
cd mcptool创建 uv 虚拟环境：
注意，确保安装 3.10 及以上版本的 Python，否则无法使用 mcp！
uv venv --python=python3.10 #mcp库依赖Python版本要大于3.10
.venv\Scripts\activate #激活这个虚拟环境uv 环境和 conda 类似，由于是一个全新的虚拟环境，所以里面的 python 库都要重新安装。类似于 pip，直接使用 uv add 来添加需要的库：
uv add mcp httpx requests创建一个执行脚本：
new-item mcptool.py如此一来，便搭建好了一个可以配置脚本的基本环境。
函数定义
接下来就来构建具体的脚本。通过设置 FastMCP 和@mcp.tool()注册mcp服务并设置异步处理函数。
mcp = FastMCP("mcptool")
@mcp.tool()通过FastMCP("mcptool")定义了当前 mcp 的服务框架名称，并通过装饰器@mcp.tool()将普通函数注册为 FastMCP 工具。
接下来，我们只需要定义一个可以异步操作的函数接口供 mcp client 使用。例如，我们声明一个 send_dify_chat_request 函数，用来接收 dify 工作流返回的结果，那么在函数声明定义处可以这么写：
async def send_dify_chat_request(
query: str,
inputs: Optional[Dict[str, Any]] = None,
user_id: str = "user123",
api_key: Optional[str] = None,
api_url: Optional[str] = None,
response_mode: str = "streaming",
timeout: int = 30
) -> Dict[str, Any]:
"""
向Dify API发送聊天请求并处理响应
Args:
query: 用户的提问内容
inputs: 允许传入App定义的各变量值, 默认为空字典
user_id: 用户标识，用于定义终端用户身份，默认为"user123"
api_key: Dify API密钥，如不提供则使用默认值
api_url: Dify API URL，如不提供则使用默认值
response_mode: 响应模式，可选"streaming"或"blocking"，默认为"streaming"
timeout: 请求超时时间（秒），默认为30
Returns:
Dict: 包含AI回答、使用情况等信息的字典
"""通过 -> Dict[str, Any]:来指定返回类型，并提供注解。下面的多行注释不仅是当前函数的运行提示，同样也是为诸如 Claude desktop 等 mcp client 工具标注当前 mcp 服务用法的注解。例如，把当前服务工具绑定到 Claude 中，可以在工具处查看这个 mcp 的用法：
之后，我们可以在当前函数中编辑主体完成脚本逻辑。我只想获得 dify 工作流的输出，相对来说操作起来比较简单，但因为 dify 自身平台的限制，如果使用阻塞式输出，即当工作流全部输出时才获取结果，很有可能会达到 dify 的超时限制从而发出错误，所以当前脚本逻辑改为流式输出，但只在结果全部输出后才返回，提升系统的稳定性。
当前函数逻辑比较简单，并没有把 api-key 单独封装，而是直接预置在脚本里，未来为了安全性应当把 api-key 设置成变量单独存储。
if api_key is None:
api_key = "你的dify API密钥" # 使用默认API密钥
if api_url is None:
api_url = "http://127.0.0.1/v1/chat-messages" #本地部署的dify访问端口设置发送到 dify 工作流的内容，以及设定输出格式：
data = {
"inputs": inputs,
"query": query,
"response_mode": response_mode,
"user": user_id
}
result = {
"success": False,
"answer": "",
"usage_info": None,
"error": None,
"workflow_info": {}
}返回的格式如下：
{
"success": Boolean, # 请求是否成功
"answer": String, # Dify API 返回的完整回答文本
"usage_info": Object, # 使用情况统计（令牌使用量等）
"error": String, # 如果发生错误，这里会包含错误信息
"workflow_info": { # 工作流相关信息
"id": String, # 工作流ID
"status": String # 工作流状态（如 "succeeded"、"failed" 等）
}
}之后加上你想要的逻辑就可以了。
最后，声明函数入口：
if __name__ == "__main__":
mcp.run(transport='stdio')通过mcp.run(transport='stdio')激活FastMCP服务实例的运行时环境，并建立基于STDIO协议的通信通道。
完整程序
import requests
import json
import sys
from typing import Any, Optional, Dict
from requests.exceptions import RequestException, Timeout, ConnectionError
from mcp.server.fastmcp import FastMCP
# 初始化FastMCP服务器
mcp = FastMCP("mcptool")
@mcp.tool()
async def send_dify_chat_request(
query: str,
inputs: Optional[Dict[str, Any]] = None,
user_id: str = "user123",
api_key: Optional[str] = None,
api_url: Optional[str] = None,
response_mode: str = "streaming",
timeout: int = 30
) -> Dict[str, Any]:
"""
向Dify API发送聊天请求并处理响应
Args:
query: 用户的提问内容
inputs: 允许传入App定义的各变量值, 默认为空字典
user_id: 用户标识，用于定义终端用户身份，默认为"user123"
api_key: Dify API密钥，如不提供则使用默认值
api_url: Dify API URL，如不提供则使用默认值
response_mode: 响应模式，可选"streaming"或"blocking"，默认为"streaming"
timeout: 请求超时时间（秒），默认为30
Returns:
Dict: 包含AI回答、使用情况等信息的字典
"""
# 设置API密钥和URL
if api_key is None:
api_key = "你的API key" # 使用默认API密钥
if api_url is None:
api_url = "http://127.0.0.1/v1/chat-messages" # 使用默认URL
# 打印调试信息
print(f"向Dify API发送请求: 查询={query}, 用户={user_id}")
# 设置请求头
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# 设置请求体
if inputs is None:
inputs = {}
data = {
"inputs": inputs,
"query": query,
"response_mode": response_mode,
"user": user_id
}
result = {
"success": False,
"answer": "",
"usage_info": None,
"error": None,
"workflow_info": {}
}
try:
# 发送请求，添加超时设置
print(f"发送请求到 {api_url}")
response = requests.post(api_url, headers=headers, json=data, stream=True, timeout=timeout)
# 处理流式响应，但只在全部接收后才输出
if response.status_code == 200:
full_answer = ""
usage_info = None
workflow_info = {}
error_occurred = False
try:
for line in response.iter_lines():
if line:
line = line.decode('utf-8')
if line.startswith('data: '):
data_str = line[6:] # 去掉 'data: ' 前缀
try:
data_json = json.loads(data_str)
event_type = data_json.get("event")
# 处理不同类型的事件
if event_type == "message":
full_answer += data_json.get("answer", "")
# 在消息结束时获取使用信息
elif event_type == "message_end":
if "metadata" in data_json:
usage_info = data_json.get("metadata", {}).get("usage")
# 处理错误事件
elif event_type == "error":
error_occurred = True
error_message = data_json.get("error", "未知错误")
print(f"API错误: {error_message}")
result["error"] = f"API错误: {error_message}"
break
# 处理工作流事件
elif event_type == "workflow_started":
workflow_info["id"] = data_json.get("workflow_run_id")
print(f"工作流开始: {workflow_info['id']}")
elif event_type == "workflow_finished":
status = data_json.get("data", {}).get("status")
print(f"工作流完成: {status}")
workflow_info["status"] = status
# 检查工作流是否失败
if status == "failed":
error_msg = data_json.get("data", {}).get("error", "未知错误")
print(f"工作流执行失败: {error_msg}")
result["error"] = f"工作流执行失败: {error_msg}"
# 处理节点事件
elif event_type in ["node_started", "node_finished"]:
node_id = data_json.get("data", {}).get("node_id")
node_status = data_json.get("data", {}).get("status", "running")
if event_type == "node_finished" and node_status == "failed":
error_msg = data_json.get("data", {}).get("error", "未知错误")
print(f"节点 {node_id} 执行失败: {error_msg}")
result["error"] = f"节点 {node_id} 执行失败: {error_msg}"
# 处理文件消息
elif event_type == "message_file":
file_id = data_json.get("id")
file_type = data_json.get("type")
file_url = data_json.get("url")
print(f"收到文件: ID={file_id}, 类型={file_type}, URL={file_url}")
if "files" not in result:
result["files"] = []
result["files"].append({
"id": file_id,
"type": file_type,
"url": file_url
})
# 处理TTS（文字转语音）消息
elif event_type in ["tts_message", "tts_message_end"]:
print(f"收到TTS事件: {event_type}")
# 处理消息替换事件
elif event_type == "message_replace":
full_answer = data_json.get("answer", "")
print("消息内容被替换（可能是由于内容审查）")
except json.JSONDecodeError as e:
error_message = f"JSON解析错误: {e}"
print(error_message)
result["error"] = error_message
except Exception as e:
error_message = f"处理消息时出错: {e}"
print(error_message)
result["error"] = error_message
except Exception as e:
error_message = f"读取响应流时出错: {e}"
print(error_message)
result["error"] = error_message
# 全部收集完后，只有在没有错误时才输出结果
if not error_occurred:
print("AI回答:", full_answer)
# 如果需要，还可以输出使用情况
if usage_info:
print("\n使用情况:")
print(f"提示令牌: {usage_info.get('prompt_tokens', 'N/A')}")
print(f"完成令牌: {usage_info.get('completion_tokens', 'N/A')}")
print(f"总令牌: {usage_info.get('total_tokens', 'N/A')}")
# 设置返回结果
result["success"] = not error_occurred
result["answer"] = full_answer
result["usage_info"] = usage_info
result["workflow_info"] = workflow_info
else:
# 处理HTTP错误代码
error_message = "未知错误"
try:
error_data = response.json()
error_message = error_data.get("error", {}).get("message", str(error_data))
except (json.JSONDecodeError, ValueError, KeyError):
error_message = response.text
print(f"请求失败，状态码: {response.status_code}")
print(f"错误信息: {error_message}")
result["error"] = f"请求失败，状态码: {response.status_code}, 错误信息: {error_message}"
# 处理常见的HTTP错误
if response.status_code == 401:
print("认证失败: 请检查您的API密钥是否正确")
result["error_type"] = "authentication_error"
elif response.status_code == 403:
print("权限不足: 您的API密钥没有足够的权限执行此操作")
result["error_type"] = "permission_error"
elif response.status_code == 404:
print("资源不存在: 请检查API端点URL是否正确")
result["error_type"] = "not_found_error"
elif response.status_code == 429:
print("请求过多: 已超过API速率限制，请稍后再试")
result["error_type"] = "rate_limit_error"
elif 500 <= response.status_code < 600:
print("服务器错误: 服务器端发生错误，请稍后再试")
result["error_type"] = "server_error"
except Timeout:
error_message = "请求超时: 服务器没有在预期的时间内响应"
print(error_message)
result["error"] = error_message
result["error_type"] = "timeout_error"
except ConnectionError:
error_message = "连接错误: 无法连接到服务器，请检查网络连接和服务器状态"
print(error_message)
result["error"] = error_message
result["error_type"] = "connection_error"
except RequestException as e:
error_message = f"请求异常: {e}"
print(error_message)
result["error"] = error_message
result["error_type"] = "request_error"
except Exception as e:
error_message = f"发生未预期的错误: {e}"
print(error_message)
result["error"] = error_message
result["error_type"] = "unexpected_error"
return result
# 直接运行时的示例用法
if __name__ == "__main__":
# import argparse
# # 检查是否需要通过命令行运行或作为MCP服务器运行
# if len(sys.argv) > 1 and sys.argv[1] == "--server":
# print("启动MCP服务器...")
# # 启动MCP服务器
# mcp.run(transport='stdio')
# else:
# # 作为命令行工具运行
# parser = argparse.ArgumentParser(description='向Dify API发送聊天请求')
# parser.add_argument('--query', type=str, help='用户的提问内容', default="gitbook是什么？怎么用？")
# parser.add_argument('--user', type=str, help='用户标识', default="user123")
# parser.add_argument('--input', type=str, help='输入变量，格式为JSON字符串', default="{}")
# parser.add_argument('--api-key', type=str, help='Dify API密钥')
# parser.add_argument('--api-url', type=str, help='Dify API URL')
# parser.add_argument('--server', action='store_true', help='以MCP服务器模式运行')
# args = parser.parse_args()
# if args.server:
# print("启动MCP服务器...")
# # 启动MCP服务器
# mcp.run(transport='stdio')
# else:
# # 处理输入参数
# try:
# inputs = json.loads(args.input)
# except json.JSONDecodeError:
# print("输入参数JSON格式错误，使用空字典作为默认值")
# inputs = {}
# # 同步调用异步函数
# import asyncio
# # 发送请求
# result = asyncio.run(send_dify_chat_request(
# query=args.query,
# inputs=inputs,
# user_id=args.user,
# api_key=args.api_key,
# api_url=args.api_url
# ))
# # 打印结果摘要
# if result["success"]:
# if not result.get("answer"):
# print("请求成功，但没有获取到回答")
# # 注意：实际回答已经在函数内部打印出来了
# else:
# print(f"请求失败: {result.get('error', '未知错误')}")
mcp.run(transport='stdio')放入 Claude 桌面应用
在 claude_desktop_config.josn 中加入：
"mcptool":{
"command":"uv",
"args":[
"--directory",
"D:\\Desktop\\Claude-Files\\mcptool", <---改为你自己存放脚本的位置
"run",
"mcptool.py"
]
}重启客户端就可以了。
实验结果：
在 chatwise 中使用
能在 Claude 桌面端使用就能在 chatwise 等带有 mcp client 功能的软件中使用，只需要改一下命令：
uv --directory D:\Desktop\Claude-Files\mcptool run mcptool.py指定你的脚本位置以及运行方法就 OK 了。
在 Cherry studio 中使用
Cherry studio 使用 mcp 总是没有 Claude 和 chatwise 丝滑，但也能用，同样把上面的单独指令放入。
