トリガープラグイン

├── _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

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 the signature/HMAC using the properties stored in subscription.properties when the subscription was created           - Verify the request is from the 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")

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 as 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 Subscription.properties and Subscription.expires_at for the next 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 the next 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)

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.        Implementing it is optional, which is why it's not an abstract method.        """        raise NotImplementedError(            "This plugin should implement `_fetch_parameter_options` method to enable dynamic select parameter"        )