mtools-dify_platform

项目结构

部署到dify平台的步骤

1.tools.yaml文件是整个工具类，例如videodeal工具类的dify配置文件

2.text_to_video_con.yaml是单项功能例如text2video功能的dify配置文件

注意这里text_to_video_con.py只是起名一样，provider中的con是服务调用，具体内容后面介绍。

3.dify平台对yaml配置文件的说明文档如下，options配置参考上图，注意部分参数只能使用固定的几个，比如参数类型、表单类型、required之类的参数。

4.配置manifest.yaml(非必须)，其中name需要小写

5.配置和dify平台连接的key、host、port

6.修改provider文件夹中tools/xxxcon.py的内容

这里invoke函数是dify平台流程调用后会调用的函数

import logging
import httpx
from typing import Generator, Any
from dify_plugin import Tool
from dify_plugin.entities.tool import ToolInvokeMessage

logger = logging.getLogger(__name__)

class TextToVideoTool(Tool):
    def _invoke(self, tool_parameters: dict[str, Any]) -> ToolInvokeMessage:
        print(tool_parameters)
        data = {
            "content": tool_parameters["content"],
            # "url": tool_parameters["url"],
        }

        # 发送请求
        with (httpx.Client(timeout=240) as client):
            print(self.runtime.credentials['DA_TOOLS_SERVICE_URL'])
            # print(data)
            response = client.post('http://' + self.runtime.credentials['DA_TOOLS_SERVICE_URL'] + '/md_processing/md_to_pdf', json=data)
            if response.status_code == 200:
                if "application/json" in response.headers.get("Content-Type", ""):
                    try:
                        json_response = response.json()
                        if json_response.get('code') == 0:
                            # 如果 JSON 中的 code 为 0，表示成功
                            print("JSON response:", json_response)
                            return self.create_json_message(json_response)
                        else:
                            print("Error code in JSON response:", json_response.get('code'))
                    except ValueError as e:
                        print("Error parsing JSON:", e)
                    # 如果响应是 PDF 或二进制文件
                elif "pdf" in response.headers.get("Content-Type", "") or "octet-stream" in response.headers.get("Content-Type", ""):
                    # 创建一个生成器，逐块返回文件流
                    file_blob = b""  # 初始化一个空的 bytes 对象
                    for chunk in response.iter_bytes(chunk_size=1024 * 1024):  # 每次读取1MB
                        file_blob += chunk  # 将文件流逐块拼接
                    response.close()  # 确保关闭响应流
                    # print(file_blob)
                    # 创建文件元数据
                    meta = {
                        "mime_type": "application/pdf"
                    }
                    # 返回文件的原始数据
                    yield self.create_text_message("转换完成")
                    yield self.create_blob_message(blob=file_blob, meta=meta)
                else:
                    raise ValueError("返回的内容类型不是pdf")
            else:
                raise ValueError(f"访问失败，状态码：{response.status_code}")