dify-docs/plugin-dev-zh/0222-trigger-plugin.mdx

---
title: "触发器插件"
---

## 触发器插件是什么？

Dify v1.10.0 引入了触发器（Trigger）类型的开始节点。区别于代码（Code）、工具（Tool）、知识检索（Knowledge Retrieval）等功能性节点，触发器的作用是**将第三方事件转化为 Dify 能接受的入参格式**。

![Trigger Plugin Intro](/images/trigger_plugin_intro.PNG)

例如，在 Gmail 中将事件的接收方配置为 Dify 后，每当你收到一封新邮件，Gmail 就会向 Dify 发送一个事件，可用于触发工作流。然而：

- Gmail 的原始请求格式不是 Dify 可接受的格式；

- 全世界有成千上万的平台，每个平台的格式都不尽相同。

因此，我们需要触发器插件来定义和解析这些来自不同平台、不同格式的事件，并将它们统一为 Dify 可接受的标准入参格式。

## 技术原理

Dify 触发器基于 Webhook 实现。Webhook 是互联网中广泛应用的机制，主流 SaaS 平台（如 Github、Slack、Linear 等）均支持 Webhook 并配有详细的开发文档。

Webhook 可理解为基于 HTTP 协议的事件派发中心。**当配置好事件接收地址后，这些 SaaS 平台会在指定事件发生时自动将其推送至目标服务器。**

为了统一处理来自不同平台的 Webhook 事件，Dify 抽象出两个核心概念：订阅（Subscription） 和事件（Event）。

- **订阅**：如前所述，基于 Webhook 的事件派发有一个前提条件——**需要将 Dify 的网络地址作为目标服务器，配置在第三方平台的开发者后台中。这个配置过程在 Dify 中称为「订阅」。**

- **事件**：一个平台可能会发送多种类型的事件，例如 *新邮件*、*邮件删除*、*邮件已读* 等，这些事件均会被推送至预先配置的地址。一个触发器插件可配置多个事件，一个事件对应 Dify 工作流中的一个触发器节点。

## 插件开发

触发器插件的开发方式与工具（Tool）、数据源（Data Source）、模型（Model）等插件类型一致，均可通过 `dify plugin init` 命令创建开发模板，其文件结构遵循统一的插件格式规范：

```
├── _assets
│   └── icon.svg
├── events
│   └── star
│       ├── star_created.py
│       └── star_created.yaml
├── main.py
├── manifest.yaml
├── provider
│   ├── github.py
│   └── github.yaml
├── README.md
├── PRIVACY.md
└── requirements.txt
```
- `manifest.yaml`：描述插件的基本信息。

- `provider` 目录：包含插件供应商的描述信息、创建订阅的代码，以及接收 Webhook 请求后对事件进行分类的代码。

- **`events` 目录：包含事件处理和筛选功能的代码，支持在节点端对事件进行本地筛选；每个目录下可创建二级目录，以便对事件进行分组。**

<Note>
    在触发器插件中，最低 Dify 版本需设置为 `1.10.0`，SDK 版本需设置为大于等于 `0.6.0`。
</Note>

下面以 GitHub 为例，介绍触发器插件的具体开发方法。

### 订阅创建

在主流 SaaS 平台中，Webhook 的配置方式差异较大：

- 部分平台（如 GitHub）支持通过 API 配置 Webhook。此类平台在 Dify 中完成 OAuth 鉴权后，Dify 即可自动完成 Webhook 配置。

- 另一类平台（如 Notion）不仅不提供 Webhook API 配置功能，还要求用户手动完成部分校验工作。

因此，我们将订阅过程分为两个部分：Subscription Constructor 和 Subscription。

对于 Notion 这类平台，订阅的创建需要用户手动将 Dify 提供的回调地址（Callback URL）复制粘贴到 Notion 后台中，以完成 Webhook 配置。此流程对应 Dify 界面中的 **粘贴 URL 以创建新订阅** 选项。

<img src="/images/trigger_plugin_manual_webhook_setup.PNG" alt="Paste URL to Create a New Subscription" width="563" />

若要实现通过手动复制 Callback URL 的方式创建订阅，需要修改 `github.yaml` 和 `github.py` 两个文件。

<Tabs>
    <Tab title="github.yaml">
        由于 Github Webhook 存在加密机制，解密和校验请求需要使用密钥，因此需在 `github.yaml` 中声明 `webhook_secret`。

        ```YAML
        subscription_schema:
        - name: "webhook_secret"
          type: "secret-input"
          required: false
          label:
            zh_Hans: "Webhook Secret"
            en_US: "Webhook Secret"
            ja_JP: "Webhookシークレット"
          help:
            en_US: "Optional webhook secret for validating GitHub webhook requests"
            ja_JP: "GitHub Webhookリクエストの検証用のオプションのWebhookシークレット"
            zh_Hans: "可选的用于验证 GitHub webhook 请求的 webhook 密钥"
        ```
    </Tab>
    <Tab title="github.py">

       首先，我们需要编写 `dispatch_event` 接口。所有发送到 Callback URL 的请求都会先经过该接口处理，处理后的事件将显示在界面上的 **请求日志** 中，便于调试与验证。

       <img src="/images/trigger_plugin_manual_webhook_setup_config.PNG" alt="Manual Setup" width="500" />

        在代码中，可通过 `subscription.properties` 获取在 `github.yaml` 中声明的 `webhook_secret`。

        `dispatch_event` 方法需要根据请求内容判断该请求对应的事件类型。在以下示例中，事件提取由 `_dispatch_trigger_event` 方法完成。

        <Tip>
            完整代码示例，请参考 [Dify GitHub 插件代码](https://github.com/langgenius/dify-plugin-sdks/tree/feat/trigger/python/examples/github_trigger)。
        </Tip>

        ```Python
        class GithubTrigger(Trigger):
            """Handle GitHub webhook event dispatch."""

            def _dispatch_event(self, subscription: Subscription, request: Request) -> EventDispatch:
                webhook_secret = subscription.properties.get("webhook_secret")
                if webhook_secret:
                    self._validate_signature(request=request, webhook_secret=webhook_secret)

                event_type: str | None = request.headers.get("X-GitHub-Event")
                if not event_type:
                    raise TriggerDispatchError("Missing GitHub event type header")

                payload: Mapping[str, Any] = self._validate_payload(request)
                response = Response(response='{"status": "ok"}', status=200, mimetype="application/json")
                event: str = self._dispatch_trigger_event(event_type=event_type, payload=payload)
                return EventDispatch(events=[event] if event else [], response=response)
        ```
    </Tab>
</Tabs>

### 事件处理

提取出事件后，需要由相应的事件实现对原始 HTTP 请求进行过滤，并将其转换为 Dify 工作流可接受的入参。

以 Issue 事件为例，可分别通过 `events/issues/issues.yaml` 和 `events/issues/issues.py` 来定义事件与实现事件。事件的输出可在 `issues.yaml` 的 `output_schema` 中定义，与工具类型插件相同，遵循 JSON Schema 规范。

<Tabs>
    <Tab title="issues.yaml">
        ```YAML
        identity:
          name: issues
          author: langgenius
          label:
            en_US: Issues
            zh_Hans: 议题
            ja_JP: イシュー
        description:
          en_US: Unified issues event with actions filter
          zh_Hans: 带 actions 过滤的统一 issues 事件
          ja_JP: アクションフィルタ付きの統合イシューイベント
        output_schema:
          type: object
          properties:
            action:
              type: string
            issue:
              type: object
              description: The issue itself
        extra:
          python:
            source: events/issues/issues.py
        ```
    </Tab>
    <Tab title="issues.py">
        ```Python
        from collections.abc import Mapping
        from typing import Any

        from werkzeug import Request

        from dify_plugin.entities.trigger import Variables
        from dify_plugin.errors.trigger import EventIgnoreError
        from dify_plugin.interfaces.trigger import Event

        class IssuesUnifiedEvent(Event):
            """Unified Issues event. Filters by actions and common issue attributes."""

            def _on_event(self, request: Request, parameters: Mapping[str, Any], payload: Mapping[str, Any]) -> Variables:
                payload = request.get_json()
                if not payload:
                    raise ValueError("No payload received")

                allowed_actions = parameters.get("actions") or []
                action = payload.get("action")
                if allowed_actions and action not in allowed_actions:
                    raise EventIgnoreError()

                issue = payload.get("issue")
                if not isinstance(issue, Mapping):
                    raise ValueError("No issue in payload")

                return Variables(variables={**payload})
        ```
    </Tab>
</Tabs>

### 事件过滤

若希望插件能够筛选掉部分事件，例如只关注具有某个特定标签的 Issue 事件，可在事件定义中为 Issue 事件添加 `parameters`。在 `_on_event` 方法中，通过抛出 `EventIgnoreError` 异常，即可在实际运行中根据配置的参数过滤掉不符合条件的事件。

<Tabs>
    <Tab title="issues.yaml">
        ```YAML
        parameters:
        - name: added_label
          label:
            en_US: Added Label
            zh_Hans: 添加的标签
            ja_JP: 追加されたラベル
          type: string
          required: false
          description:
            en_US: "Only trigger if these specific labels were added (e.g., critical, priority-high, security, comma-separated). Leave empty to trigger for any label addition."
            zh_Hans: "仅当添加了这些特定标签时触发（例如：critical, priority-high, security，逗号分隔）。留空则对任何标签添加触发。"
            ja_JP: "これらの特定のラベルが追加された場合のみトリガー（例: critical, priority-high, security，カンマ区切り）。空の場合は任意のラベル追加でトリガー。"
        ```
    </Tab>
    <Tab title="issues.py">
        ```Python
        def _check_added_label(self, payload: Mapping[str, Any], added_label_param: str | None) -> None:
            """Check if the added label matches the allowed labels"""
            if not added_label_param:
                return

            allowed_labels = [label.strip() for label in added_label_param.split(",") if label.strip()]
            if not allowed_labels:
                return

            # The payload contains the label that was added
            label = payload.get("label", {})
            label_name = label.get("name", "")

            if label_name not in allowed_labels:
                raise EventIgnoreError()
                
        def _on_event(self, request: Request, parameters: Mapping[str, Any], payload: Mapping[str, Any]) -> Variables:
            # ...
            # Apply all filters
            self._check_added_label(payload, parameters.get("added_label"))

            return Variables(variables={**payload})
        ```
    </Tab>
</Tabs>

### 通过 OAuth 或 API key 创建订阅

若要实现通过 OAuth 或 API key 自动创建订阅，同样需要修改 `github.yaml` 和 `github.py` 两个文件。

<Tabs>
    <Tab title="github.yaml">
        在 `github.yaml` 中，添加如下字段。

        ```YAML
        subscription_constructor:
          parameters:
          - name: "repository"
            label:
              en_US: "Repository"
              zh_Hans: "仓库"
              ja_JP: "リポジトリ"
            type: "dynamic-select"
            required: true
            placeholder:
              en_US: "owner/repo"
              zh_Hans: "owner/repo"
              ja_JP: "owner/repo"
            help:
              en_US: "GitHub repository in format owner/repo (e.g., microsoft/vscode)"
              zh_Hans: "GitHub 仓库，格式为 owner/repo（例如：microsoft/vscode）"
              ja_JP: "GitHubリポジトリは owner/repo 形式で入力してください（例: microsoft/vscode）"
          credentials_schema:
            access_tokens:
              help:
                en_US: Get your Access Tokens from GitHub
                ja_JP: GitHub からアクセストークンを取得してください
                zh_Hans: 从 GitHub 获取您的 Access Tokens
              label:
                en_US: Access Tokens
                ja_JP: アクセストークン
                zh_Hans: Access Tokens
              placeholder:
                en_US: Please input your GitHub Access Tokens
                ja_JP: GitHub のアクセストークンを入力してください
                zh_Hans: 请输入你的 GitHub Access Tokens
              required: true
              type: secret-input
              url: https://github.com/settings/tokens?type=beta
          extra:
            python:
              source: provider/github.py
        ```
        `subscription_constructor` 是 Dify 抽象出的一个概念，用于定义如何构建订阅，包含以下字段：

        - `parameters`（可选）：用于定义创建订阅时所需的参数，例如想要订阅的事件类型或目标 GitHub 仓库等。

        - `credentials_schema`（可选）：通过 API key 创建订阅时，必须填写该字段以声明所需的密钥，例如在 GitHub 中定义 `access_tokens`。

        - `oauth_schema`（可选）：通过 OAuth 创建订阅时，必须填写该字段以启用 OAuth 鉴权。具体定义方式，可参考 [为插件添加 OAuth 支持](/plugin-dev-zh/0222-tool-oauth)。
    </Tab>
    <Tab title="github.py">
        在 `github.py` 中创建一个 `Constructor` 类，以实现自动化订阅逻辑。

        ```Python
        class GithubSubscriptionConstructor(TriggerSubscriptionConstructor):
            """Manage GitHub trigger subscriptions."""
            def _validate_api_key(self, credentials: Mapping[str, Any]) -> None:
                # ...

            def _create_subscription(
                self,
                endpoint: str,
                parameters: Mapping[str, Any],
                credentials: Mapping[str, Any],
                credential_type: CredentialType,
            ) -> Subscription:
                repository = parameters.get("repository")
                if not repository:
                    raise ValueError("repository is required (format: owner/repo)")

                try:
                    owner, repo = repository.split("/")
                except ValueError:
                    raise ValueError("repository must be in format 'owner/repo'") from None

                events: list[str] = parameters.get("events", [])
                webhook_secret = uuid.uuid4().hex
                url = f"https://api.github.com/repos/{owner}/{repo}/hooks"
                headers = {
                    "Authorization": f"Bearer {credentials.get('access_tokens')}",
                    "Accept": "application/vnd.github+json",
                }

                webhook_data = {
                    "name": "web",
                    "active": True,
                    "events": events,
                    "config": {"url": endpoint, "content_type": "json", "insecure_ssl": "0", "secret": webhook_secret},
                }

                try:
                    response = requests.post(url, json=webhook_data, headers=headers, timeout=10)
                except requests.RequestException as exc:
                    raise SubscriptionError(f"Network error while creating webhook: {exc}", error_code="NETWORK_ERROR") from exc

                if response.status_code == 201:
                    webhook = response.json()
                    return Subscription(
                        expires_at=int(time.time()) + self._WEBHOOK_TTL,
                        endpoint=endpoint,
                        parameters=parameters,
                        properties={
                            "external_id": str(webhook["id"]),
                            "repository": repository,
                            "events": events,
                            "webhook_secret": webhook_secret,
                            "active": webhook.get("active", True),
                        },
                    )

                response_data: dict[str, Any] = response.json() if response.content else {}
                error_msg = response_data.get("message", "Unknown error")
                error_details = response_data.get("errors", [])
                detailed_error = f"Failed to create GitHub webhook: {error_msg}"
                if error_details:
                    detailed_error += f" Details: {error_details}"

                raise SubscriptionError(
                    detailed_error,
                    error_code="WEBHOOK_CREATION_FAILED",
                    external_response=response_data,
                )
        ```
    </Tab>
</Tabs>

---

完成以上两个文件的配置后，你将在 Dify 界面中看到 **通过 API Key 创建** 选项。

通过 OAuth 的订阅自动创建也可以在 `Constructor` 中实现：在 `subscription_constructor` 中添加 `oauth_schema` 字段，即可启用 OAuth 认证方式。

![OAuth & API Key Options](/images/trigger_plugin_oauth_apikey.png)

## 探索更多

触发器插件开发中核心类的接口定义和实现方法如下。

### Trigger

```Python
class Trigger(ABC):
    @abstractmethod
    def _dispatch_event(self, subscription: Subscription, request: Request) -> EventDispatch:
        """
        Internal method to implement event dispatch logic.

        Subclasses must override this method to handle incoming webhook events.

        Implementation checklist:
        1. Validate the webhook request:
           - Check signature/HMAC using properties when you create the subscription from subscription.properties
           - Verify request is from expected source
        2. Extract event information:
           - Parse event type from headers or body
           - Extract relevant payload data
        3. Return EventDispatch with:
           - events: List of Event names to invoke (can be single or multiple)
           - response: Appropriate HTTP response for the webhook

        Args:
            subscription: The Subscription object with endpoint and properties fields
            request: Incoming webhook HTTP request

        Returns:
            EventDispatch: Event dispatch routing information

        Raises:
            TriggerValidationError: For security validation failures
            TriggerDispatchError: For parsing or routing errors
        """
        raise NotImplementedError("This plugin should implement `_dispatch_event` method to enable event dispatch")

```

### TriggerSubscriptionConstructor

```Python
class TriggerSubscriptionConstructor(ABC, OAuthProviderProtocol):
    # OPTIONAL
    def _validate_api_key(self, credentials: Mapping[str, Any]) -> None:
        raise NotImplementedError(
            "This plugin should implement `_validate_api_key` method to enable credentials validation"
        )
        
    # OPTIONAL
    def _oauth_get_authorization_url(self, redirect_uri: str, system_credentials: Mapping[str, Any]) -> str:
        raise NotImplementedError(
            "The trigger you are using does not support OAuth, please implement `_oauth_get_authorization_url` method"
        )
    
    # OPTIONAL
    def _oauth_get_credentials(
        self, redirect_uri: str, system_credentials: Mapping[str, Any], request: Request
    ) -> TriggerOAuthCredentials:
        raise NotImplementedError(
            "The trigger you are using does not support OAuth, please implement `_oauth_get_credentials` method"
        )
    
    # OPTIONAL
    def _oauth_refresh_credentials(
        self, redirect_uri: str, system_credentials: Mapping[str, Any], credentials: Mapping[str, Any]
    ) -> OAuthCredentials:
        raise NotImplementedError(
            "The trigger you are using does not support OAuth, please implement `_oauth_refresh_credentials` method"
        )

    @abstractmethod
    def _create_subscription(
        self,
        endpoint: str,
        parameters: Mapping[str, Any],
        credentials: Mapping[str, Any],
        credential_type: CredentialType,
    ) -> Subscription:
        """
        Internal method to implement subscription logic.

        Subclasses must override this method to handle subscription creation.

        Implementation checklist:
        1. Use the endpoint parameter provided by Dify
        2. Register webhook with external service using their API
        3. Store all necessary information in Subscription.properties for future operations(e.g., dispatch_event)
        4. Return Subscription with:
           - expires_at: Set appropriate expiration time
           - endpoint: The webhook endpoint URL allocated by Dify for receiving events, same with the endpoint parameter
           - parameters: The parameters of the subscription
           - properties: All configuration and external IDs

        Args:
            endpoint: The webhook endpoint URL allocated by Dify for receiving events
            parameters: Subscription creation parameters
            credentials: Authentication credentials
            credential_type: The type of the credentials, e.g., "api-key", "oauth2", "unauthorized"

        Returns:
            Subscription: Subscription details with metadata for future operations

        Raises:
            SubscriptionError: For operational failures (API errors, invalid credentials)
            ValueError: For programming errors (missing required params)
        """
        raise NotImplementedError(
            "This plugin should implement `_create_subscription` method to enable event subscription"
        )

    @abstractmethod
    def _delete_subscription(
        self, subscription: Subscription, credentials: Mapping[str, Any], credential_type: CredentialType
    ) -> UnsubscribeResult:
        """
        Internal method to implement unsubscription logic.

        Subclasses must override this method to handle subscription removal.

        Implementation guidelines:
        1. Extract necessary IDs from subscription.properties (e.g., external_id)
        2. Use credentials and credential_type to call external service API to delete the webhook
        3. Handle common errors (not found, unauthorized, etc.)
        4. Always return UnsubscribeResult with detailed status
        5. Never raise exceptions for operational failures - use UnsubscribeResult.success=False

        Args:
            subscription: The Subscription object with endpoint and properties fields

        Returns:
            UnsubscribeResult: Always returns result, never raises for operational failures
        """
        raise NotImplementedError(
            "This plugin should implement `_delete_subscription` method to enable event unsubscription"
        )

    @abstractmethod
    def _refresh_subscription(
        self, subscription: Subscription, credentials: Mapping[str, Any], credential_type: CredentialType
    ) -> Subscription:
        """
        Internal method to implement subscription refresh logic.

        Subclasses must override this method to handle simple expiration extension.

        Implementation patterns:
        1. For webhooks without expiration (e.g., GitHub):
           - Update the Subscription.expires_at=-1 then Dify will never call this method again

        2. For lease-based subscriptions (e.g., Microsoft Graph):
           - Use the information in Subscription.properties to call service's lease renewal API if available
           - Handle renewal limits (some services limit renewal count)
           - Update the Subscription.properties and Subscription.expires_at for next time renewal if needed

        Args:
            subscription: Current subscription with properties
            credential_type: The type of the credentials, e.g., "api-key", "oauth2", "unauthorized"
            credentials: Current authentication credentials from credentials_schema.
                        For API key auth, according to `credentials_schema` defined in the YAML.
                        For OAuth auth, according to `oauth_schema.credentials_schema` defined in the YAML.
                        For unauthorized auth, there is no credentials.

        Returns:
            Subscription: Same subscription with extended expiration
                        or new properties and expires_at for next time renewal

        Raises:
            SubscriptionError: For operational failures (API errors, invalid credentials)
        """
        raise NotImplementedError("This plugin should implement `_refresh` method to enable subscription refresh")
    
    # OPTIONAL
    def _fetch_parameter_options(
        self, parameter: str, credentials: Mapping[str, Any], credential_type: CredentialType
    ) -> list[ParameterOption]:
        """
        Fetch the parameter options of the trigger.

        Implementation guidelines:
        When you need to fetch parameter options from an external service, use the credentials
        and credential_type to call the external service API, then return the options to Dify
        for user selection.

        Args:
            parameter: The parameter name for which to fetch options
            credentials: Authentication credentials for the external service
            credential_type: The type of credentials (e.g., "api-key", "oauth2", "unauthorized")

        Returns:
            list[ParameterOption]: A list of available options for the parameter

        Examples:
            GitHub Repositories:
            >>> result = provider.fetch_parameter_options(parameter="repository")
            >>> print(result)  # [ParameterOption(label="owner/repo", value="owner/repo")]

            Slack Channels:
            >>> result = provider.fetch_parameter_options(parameter="channel")
            >>> print(result)
```

### Event

```Python
class Event(ABC):
    @abstractmethod
    def _on_event(self, request: Request, parameters: Mapping[str, Any], payload: Mapping[str, Any]) -> Variables:
        """
        Transform the incoming webhook request into structured Variables.

        This method should:
        1. Parse the webhook payload from the request
        2. Apply filtering logic based on parameters
        3. Extract relevant data matching the output_schema
        4. Return a structured Variables object

        Args:
            request: The incoming webhook HTTP request containing the raw payload.
                    Use request.get_json() to parse JSON body.
            parameters: User-configured parameters for filtering and transformation
                       (e.g., label filters, regex patterns, threshold values).
                       These come from the subscription configuration.
            payload: The decoded payload from previous step `Trigger.dispatch_event`.
                     It will be delivered into `_on_event` method.
        Returns:
            Variables: Structured variables matching the output_schema
                      defined in the event's YAML configuration.

        Raises:
            EventIgnoreError: When the event should be filtered out based on parameters
            ValueError: When the payload is invalid or missing required fields

        Example:
            >>> def _on_event(self, request, parameters):
            ...     payload = request.get_json()
            ...
            ...     # Apply filters
            ...     if not self._matches_filters(payload, parameters):
            ...         raise EventIgnoreError()
            ...
            ...     # Transform data
            ...     return Variables(variables={
            ...         "title": payload["issue"]["title"],
            ...         "author": payload["issue"]["user"]["login"],
            ...         "url": payload["issue"]["html_url"],
            ...     })
        """

    def _fetch_parameter_options(self, parameter: str) -> list[ParameterOption]:
        """
        Fetch the parameter options of the trigger.

        To be implemented by subclasses.

        Also, it's optional to implement, that's why it's not an abstract method.
        """
        raise NotImplementedError(
            "This plugin should implement `_fetch_parameter_options` method to enable dynamic select parameter"
        )
```

{/*
Contributing Section
DO NOT edit this section!
It will be automatically generated by the script.
*/}

---

[编辑此页面](https://github.com/langgenius/dify-docs/edit/main/plugin-dev-zh/0222-trigger-plugin.mdx) | [提交问题](https://github.com/langgenius/dify-docs/issues/new?template=docs.yml)