Dify 本地 tools 集成指南

方式一：

采用 本地 API 注册 的模式

本地构架 restful API, 满足OpenAPI 的Schema 文件，

具体协议可以参考：https://swagger.io/specification/

本地实现

from fastapi import FastAPI
import uvicorn
import socket
#需要实现
from weather_service import WeatherService 

host = socket.gethostbyname(socket.gethostname())

# 指定 servers 参数
servers = [
    {
        "url": f"http://{host}:8000",
        "description": "本地开发服务器"
    }
]

app = FastAPI(servers=servers)
weather_service = WeatherService()

@app.get("/weather/{city}", summary="获取城市当前天气", description="根据城市名称获取当前天气信息", operation_id="custom_get_weather")
def get_weather(city: str):
    """
    获取指定城市的当前天气信息。

    - **city**: 城市名称
    """
    return weather_service.get_city_weather(city)

@app.get("/air-quality/{city}", summary="获取城市当前空气质量", description="根据城市名称获取当前空气质量信息", operation_id="custom_get_air_quality")
def get_air_quality(city: str):
    """
    获取指定城市的当前空气质量信息。

    - **city**: 城市名称
    """
    return weather_service.get_current_air_quality(city)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

备注：FastAPI会自动生成

生成˙的

• swagger 文档：http://localhost:8000/docs#/

• openapi 文件：http://localhost:8000/openapi.json

平台中配置

记得地址要改成本机IP

从文字可以看到，目前的本地的工具 已经可以被配置到了平台中，后续再平台的工作流，agent 中均可以使用

平台中使用

在聊天中我们可以看到，调用工具的具体信息

方式二：

采用 MCP 的模式

官方的架构图

目前 MCP 官方支持2种传输实现：

Standard Input/Output (stdio) 标准输入输出

• Building command-line tools

• Implementing local integrations

• Needing simple process communication

• Working with shell scripts

目前这种传输实现主要是为了 本地的资源集成和命令行工具，从我们目前本地 操作 API 去集成平台的目的分析，此方式是不可用~

Server-Sent Events (SSE) 服务器发送事件

SSE 传输通过 HTTP POST 请求实现服务器到客户端的流式通信。

在以下情况下使用 SSE：

• 仅需要 server-to-client 的流式通信

• 在受限网络中工作

• 实现简单更新

备注：此方式会对服务器有一定要压力，主要是长连接，主要是让客户端和服务器保持持续的会话，这样服务器可以随时推送。但对大规模分布式部署可能不太友好，MCP 的社区已经在做改进，预计后期会采用可流式 HTTP 传输，让服务器和客户端可以在需要时重连并恢复会话，既能做到状态追踪，也更便于横向扩容或处理断线情况。

本地Service代码实现：

from mcp.server.fastmcp import FastMCP
from weather_service import WeatherService

mcp = FastMCP("weather_sse", port=8002, host="0.0.0.0")
weather_service = WeatherService()

@mcp.tool()
async def get_weather(city: str) -> str:
    """
     获取指定城市的当前天气信息。
    参数：
        - **city**: 城市名称
    返回：
        str: 天气信息。
    """
    return weather_service.get_city_weather(city)

@mcp.tool()
async def get_air_quality(city: str):
    """
    获取指定城市的当前空气质量信息。

    - **city**: 城市名称
    """
    return weather_service.get_current_air_quality(city)

if __name__ == "__main__":
    mcp.run(transport="sse")

具体的 mcp的安装流程，可以参考官方的 quickstart- service模块的开发

平台的配置

在平台的应用市场中，找到 MCP SSE 的插件并安装

安装后再工作中，选择此工具进行配置，配置格式如下：

{
    "server_name": {
        "url": "http://{本地IP}:8002/sse",
        "headers": {},
        "timeout": 60,
        "sse_read_timeout": 3000
    }
}

平台的使用