第 20 期：自定义工具开发

自定义工具架构

graph LR
    Agent[Dify Agent] --> ToolNode[Tool 节点]
    ToolNode --> Custom[自定义工具]
    Custom --> |OpenAPI Schema| ExtAPI[你的外部 API]

方式一：OpenAPI Schema 接入

将你已有的 REST API 通过 OpenAPI (Swagger) 规范接入 Dify。

# openapi.yaml - 自定义库存查询工具
openapi: "3.0.0"
info:
  title: "库存管理 API"
  version: "1.0.0"
servers:
  - url: "https://api.your-company.com/v1"
paths:
  /inventory/search:
    get:
      operationId: searchInventory
      summary: "搜索产品库存信息"
      description: "根据产品名称或 SKU 查询实时库存数量和仓库位置"
      parameters:
        - name: keyword
          in: query
          required: true
          schema:
            type: string
          description: "产品名称或 SKU 编号"
        - name: warehouse
          in: query
          required: false
          schema:
            type: string
            enum: ["上海", "北京", "广州"]
          description: "指定仓库"
      responses:
        "200":
          description: "查询成功"
          content:
            application/json:
              schema:
                type: object
                properties:
                  products:
                    type: array
                    items:
                      type: object
                      properties:
                        name:
                          type: string
                        sku:
                          type: string
                        quantity:
                          type: integer
                        warehouse:
                          type: string

在 Dify 后台 → Workspace → Tools → Custom Tool → Import from URL/File，上传这个 YAML 即可。

方式二：编写 Tool Provider (Plugin)

# 自定义工具 Provider 示例
# tools/weather_provider.py

from typing import Any
from dify_plugin import ToolProvider, Tool

class WeatherProvider(ToolProvider):
    """天气查询工具提供商"""
    
    def validate_credentials(self, credentials: dict) -> None:
        """验证 API 密钥是否有效"""
        api_key = credentials.get("api_key")
        if not api_key:
            raise ValueError("API Key 不能为空")

class GetWeatherTool(Tool):
    """获取天气信息"""
    
    def _invoke(self, tool_parameters: dict) -> Any:
        city = tool_parameters["city"]
        api_key = self.runtime.credentials["api_key"]
        
        import requests
        response = requests.get(
            "https://api.weatherapi.com/v1/current.json",
            params={"key": api_key, "q": city, "lang": "zh"}
        )
        data = response.json()
        
        return {
            "city": data["location"]["name"],
            "temperature": data["current"]["temp_c"],
            "condition": data["current"]["condition"]["text"],
            "humidity": data["current"]["humidity"]
        }

工具定义文件：

# tools/weather_provider.yaml
identity:
  name: weather_provider
  author: your-name
  label:
    en_US: Weather Query
    zh_Hans: 天气查询
description:
  human:
    en_US: Query real-time weather information
    zh_Hans: 查询实时天气信息
credentials:
  - name: api_key
    type: secret
    required: true
    label:
      en_US: API Key
tools:
  - name: get_weather
    label:
      zh_Hans: 获取天气
    parameters:
      - name: city
        type: string
        required: true
        label:
          zh_Hans: 城市名称

调试技巧

# 测试自定义工具
def test_custom_tool():
    """本地测试工具逻辑"""
    tool = GetWeatherTool()
    tool.runtime = MockRuntime(credentials={"api_key": "test-key"})
    
    result = tool._invoke({"city": "上海"})
    
    assert "temperature" in result
    assert "condition" in result
    print(f"✅ 测试通过: {result}")

test_custom_tool()