Merge remote-tracking branch 'origin/feat/trigger' into feat/trigger

- Added validation for `PluginTriggerData` and `ScheduleTriggerData` in the `WorkflowService` to support new trigger types.
- Updated debug event return strings in `PluginTriggerDebugEvent` and `WebhookDebugEvent` for clarity and consistency.
- Enhanced logging in `dispatch_triggered_workflows_async` to include subscription and provider IDs, improving traceability during trigger dispatching.

.devcontainer/post_create_command.sh

@@ -1,6 +1,7 @@
 #!/bin/bash
 WORKSPACE_ROOT=$(pwd)
 
+npm add -g pnpm@10.15.0
 corepack enable
 cd web && pnpm install
 pipx install uv

.github/workflows/autofix.yml

@@ -2,6 +2,8 @@ name: autofix.ci
 on:
   pull_request:
     branches: ["main"]
+  push:
+    branches: ["main"]
 permissions:
   contents: read
 
@@ -30,6 +32,8 @@ jobs:
         run: |
           uvx --from ast-grep-cli sg --pattern 'db.session.query($WHATEVER).filter($HERE)' --rewrite 'db.session.query($WHATEVER).where($HERE)' -l py --update-all
           uvx --from ast-grep-cli sg --pattern 'session.query($WHATEVER).filter($HERE)' --rewrite 'session.query($WHATEVER).where($HERE)' -l py --update-all
+          uvx --from ast-grep-cli sg -p '$A = db.Column($$$B)' -r '$A = mapped_column($$$B)' -l py --update-all
+          uvx --from ast-grep-cli sg -p '$A : $T = db.Column($$$B)' -r '$A : $T = mapped_column($$$B)' -l py --update-all
           # Convert Optional[T] to T | None (ignoring quoted types)
           cat > /tmp/optional-rule.yml << 'EOF'
           id: convert-optional-to-union

.github/workflows/build-push.yml

@@ -4,8 +4,7 @@ on:
   push:
     branches:
       - "main"
-      - "deploy/dev"
-      - "deploy/enterprise"
+      - "deploy/**"
       - "build/**"
       - "release/e-*"
       - "hotfix/**"

.github/workflows/deploy-trigger-dev.yml

@@ -1,4 +1,4 @@
-name: Deploy RAG Dev
+name: Deploy Trigger Dev
 
 permissions:
   contents: read
@@ -7,7 +7,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "deploy/rag-dev"
+      - "deploy/trigger-dev"
     types:
       - completed
 
@@ -16,12 +16,12 @@ jobs:
     runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      github.event.workflow_run.head_branch == 'deploy/rag-dev'
+      github.event.workflow_run.head_branch == 'deploy/trigger-dev'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v0.1.8
         with:
-          host: ${{ secrets.RAG_SSH_HOST }}
+          host: ${{ secrets.TRIGGER_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.gitignore

@@ -234,4 +234,7 @@ scripts/stress-test/reports/
 
 # mcp
 .playwright-mcp/
-.serena/
+.serena/
+
+# settings
+*.local.json

api/.env.example

@@ -343,6 +343,15 @@ OCEANBASE_VECTOR_DATABASE=test
 OCEANBASE_MEMORY_LIMIT=6G
 OCEANBASE_ENABLE_HYBRID_SEARCH=false
 
+# AlibabaCloud MySQL Vector configuration
+ALIBABACLOUD_MYSQL_HOST=127.0.0.1
+ALIBABACLOUD_MYSQL_PORT=3306
+ALIBABACLOUD_MYSQL_USER=root
+ALIBABACLOUD_MYSQL_PASSWORD=root
+ALIBABACLOUD_MYSQL_DATABASE=dify
+ALIBABACLOUD_MYSQL_MAX_CONNECTION=5
+ALIBABACLOUD_MYSQL_HNSW_M=6
+
 # openGauss configuration
 OPENGAUSS_HOST=127.0.0.1
 OPENGAUSS_PORT=6600
@@ -445,6 +454,9 @@ HTTP_REQUEST_NODE_MAX_BINARY_SIZE=10485760
 HTTP_REQUEST_NODE_MAX_TEXT_SIZE=1048576
 HTTP_REQUEST_NODE_SSL_VERIFY=True
 
+# Webhook request configuration
+WEBHOOK_REQUEST_BODY_MAX_SIZE=10485760
+
 # Respect X-* headers to redirect clients
 RESPECT_XFORWARD_HEADERS_ENABLED=false
 
@@ -522,6 +534,12 @@ ENABLE_CLEAN_MESSAGES=false
 ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK=false
 ENABLE_DATASETS_QUEUE_MONITOR=false
 ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK=true
+ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK=true
+# Interval time in minutes for polling scheduled workflows(default: 1 min)
+WORKFLOW_SCHEDULE_POLLER_INTERVAL=1
+WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE=100
+# Maximum number of scheduled workflows to dispatch per tick (0 for unlimited)
+WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK=0
 
 # Position configuration
 POSITION_TOOL_PINS=

api/.vscode/launch.json.example

@@ -54,7 +54,7 @@
                 "--loglevel",
                 "DEBUG",
                 "-Q",
-                "dataset,generation,mail,ops_trace,app_deletion"
+                "dataset,generation,mail,ops_trace,app_deletion,workflow"
             ]
         }
     ]

api/commands.py

@@ -15,12 +15,12 @@ from sqlalchemy.orm import sessionmaker
 from configs import dify_config
 from constants.languages import languages
 from core.helper import encrypter
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.plugin import PluginInstaller
 from core.rag.datasource.vdb.vector_factory import Vector
 from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
 from core.rag.models.document import Document
-from core.tools.entities.tool_entities import CredentialType
 from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
@@ -1227,6 +1227,55 @@ def setup_system_tool_oauth_client(provider, client_params):
     click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
 
 
+@click.command("setup-system-trigger-oauth-client", help="Setup system trigger oauth client.")
+@click.option("--provider", prompt=True, help="Provider name")
+@click.option("--client-params", prompt=True, help="Client Params")
+def setup_system_trigger_oauth_client(provider, client_params):
+    """
+    Setup system trigger oauth client
+    """
+    from models.provider_ids import TriggerProviderID
+    from models.trigger import TriggerOAuthSystemClient
+
+    provider_id = TriggerProviderID(provider)
+    provider_name = provider_id.provider_name
+    plugin_id = provider_id.plugin_id
+
+    try:
+        # json validate
+        click.echo(click.style(f"Validating client params: {client_params}", fg="yellow"))
+        client_params_dict = TypeAdapter(dict[str, Any]).validate_json(client_params)
+        click.echo(click.style("Client params validated successfully.", fg="green"))
+
+        click.echo(click.style(f"Encrypting client params: {client_params}", fg="yellow"))
+        click.echo(click.style(f"Using SECRET_KEY: `{dify_config.SECRET_KEY}`", fg="yellow"))
+        oauth_client_params = encrypt_system_oauth_params(client_params_dict)
+        click.echo(click.style("Client params encrypted successfully.", fg="green"))
+    except Exception as e:
+        click.echo(click.style(f"Error parsing client params: {str(e)}", fg="red"))
+        return
+
+    deleted_count = (
+        db.session.query(TriggerOAuthSystemClient)
+        .filter_by(
+            provider=provider_name,
+            plugin_id=plugin_id,
+        )
+        .delete()
+    )
+    if deleted_count > 0:
+        click.echo(click.style(f"Deleted {deleted_count} existing oauth client params.", fg="yellow"))
+
+    oauth_client = TriggerOAuthSystemClient(
+        provider=provider_name,
+        plugin_id=plugin_id,
+        encrypted_oauth_params=oauth_client_params,
+    )
+    db.session.add(oauth_client)
+    db.session.commit()
+    click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
+
+
 def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
     """
     Find draft variables that reference non-existent apps.
@@ -1521,6 +1570,14 @@ def transform_datasource_credentials():
                     auth_count = 0
                     for firecrawl_tenant_credential in firecrawl_tenant_credentials:
                         auth_count += 1
+                        if not firecrawl_tenant_credential.credentials:
+                            click.echo(
+                                click.style(
+                                    f"Skipping firecrawl credential for tenant {tenant_id} due to missing credentials.",
+                                    fg="yellow",
+                                )
+                            )
+                            continue
                         # get credential api key
                         credentials_json = json.loads(firecrawl_tenant_credential.credentials)
                         api_key = credentials_json.get("config", {}).get("api_key")
@@ -1576,6 +1633,14 @@ def transform_datasource_credentials():
                     auth_count = 0
                     for jina_tenant_credential in jina_tenant_credentials:
                         auth_count += 1
+                        if not jina_tenant_credential.credentials:
+                            click.echo(
+                                click.style(
+                                    f"Skipping jina credential for tenant {tenant_id} due to missing credentials.",
+                                    fg="yellow",
+                                )
+                            )
+                            continue
                         # get credential api key
                         credentials_json = json.loads(jina_tenant_credential.credentials)
                         api_key = credentials_json.get("config", {}).get("api_key")

api/configs/feature/__init__.py

@@ -174,6 +174,17 @@ class CodeExecutionSandboxConfig(BaseSettings):
     )
 
 
+class TriggerConfig(BaseSettings):
+    """
+    Configuration for trigger
+    """
+
+    WEBHOOK_REQUEST_BODY_MAX_SIZE: PositiveInt = Field(
+        description="Maximum allowed size for webhook request bodies in bytes",
+        default=10485760,
+    )
+
+
 class PluginConfig(BaseSettings):
     """
     Plugin configs
@@ -985,6 +996,22 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable check upgradable plugin task",
         default=True,
     )
+    ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK: bool = Field(
+        description="Enable workflow schedule poller task",
+        default=True,
+    )
+    WORKFLOW_SCHEDULE_POLLER_INTERVAL: int = Field(
+        description="Workflow schedule poller interval in minutes",
+        default=1,
+    )
+    WORKFLOW_SCHEDULE_POLLER_BATCH_SIZE: int = Field(
+        description="Maximum number of schedules to process in each poll batch",
+        default=100,
+    )
+    WORKFLOW_SCHEDULE_MAX_DISPATCH_PER_TICK: int = Field(
+        description="Maximum schedules to dispatch per tick (0=unlimited, circuit breaker)",
+        default=0,
+    )
 
 
 class PositionConfig(BaseSettings):
@@ -1108,6 +1135,7 @@ class FeatureConfig(
     AuthConfig,  # Changed from OAuthConfig to AuthConfig
     BillingConfig,
     CodeExecutionSandboxConfig,
+    TriggerConfig,
     PluginConfig,
     MarketplaceConfig,
     DataSetConfig,

api/configs/middleware/__init__.py

@@ -18,6 +18,7 @@ from .storage.opendal_storage_config import OpenDALStorageConfig
 from .storage.supabase_storage_config import SupabaseStorageConfig
 from .storage.tencent_cos_storage_config import TencentCloudCOSStorageConfig
 from .storage.volcengine_tos_storage_config import VolcengineTOSStorageConfig
+from .vdb.alibabacloud_mysql_config import AlibabaCloudMySQLConfig
 from .vdb.analyticdb_config import AnalyticdbConfig
 from .vdb.baidu_vector_config import BaiduVectorDBConfig
 from .vdb.chroma_config import ChromaConfig
@@ -330,6 +331,7 @@ class MiddlewareConfig(
     ClickzettaConfig,
     HuaweiCloudConfig,
     MilvusConfig,
+    AlibabaCloudMySQLConfig,
     MyScaleConfig,
     OpenSearchConfig,
     OracleConfig,

api/configs/middleware/vdb/alibabacloud_mysql_config.py

@@ -0,0 +1,54 @@
+from pydantic import Field, PositiveInt
+from pydantic_settings import BaseSettings
+
+
+class AlibabaCloudMySQLConfig(BaseSettings):
+    """
+    Configuration settings for AlibabaCloud MySQL vector database
+    """
+
+    ALIBABACLOUD_MYSQL_HOST: str = Field(
+        description="Hostname or IP address of the AlibabaCloud MySQL server (e.g., 'localhost' or 'mysql.aliyun.com')",
+        default="localhost",
+    )
+
+    ALIBABACLOUD_MYSQL_PORT: PositiveInt = Field(
+        description="Port number on which the AlibabaCloud MySQL server is listening (default is 3306)",
+        default=3306,
+    )
+
+    ALIBABACLOUD_MYSQL_USER: str = Field(
+        description="Username for authenticating with AlibabaCloud MySQL (default is 'root')",
+        default="root",
+    )
+
+    ALIBABACLOUD_MYSQL_PASSWORD: str = Field(
+        description="Password for authenticating with AlibabaCloud MySQL (default is an empty string)",
+        default="",
+    )
+
+    ALIBABACLOUD_MYSQL_DATABASE: str = Field(
+        description="Name of the AlibabaCloud MySQL database to connect to (default is 'dify')",
+        default="dify",
+    )
+
+    ALIBABACLOUD_MYSQL_MAX_CONNECTION: PositiveInt = Field(
+        description="Maximum number of connections in the connection pool",
+        default=5,
+    )
+
+    ALIBABACLOUD_MYSQL_CHARSET: str = Field(
+        description="Character set for AlibabaCloud MySQL connection (default is 'utf8mb4')",
+        default="utf8mb4",
+    )
+
+    ALIBABACLOUD_MYSQL_DISTANCE_FUNCTION: str = Field(
+        description="Distance function used for vector similarity search in AlibabaCloud MySQL "
+        "(e.g., 'cosine', 'euclidean')",
+        default="cosine",
+    )
+
+    ALIBABACLOUD_MYSQL_HNSW_M: PositiveInt = Field(
+        description="Maximum number of connections per layer for HNSW vector index (default is 6, range: 3-200)",
+        default=6,
+    )

api/configs/middleware/vdb/opensearch_config.py

@@ -1,23 +1,24 @@
-from enum import Enum
+from enum import StrEnum
 from typing import Literal
 
 from pydantic import Field, PositiveInt
 from pydantic_settings import BaseSettings
 
 
+class AuthMethod(StrEnum):
+    """
+    Authentication method for OpenSearch
+    """
+
+    BASIC = "basic"
+    AWS_MANAGED_IAM = "aws_managed_iam"
+
+
 class OpenSearchConfig(BaseSettings):
     """
     Configuration settings for OpenSearch
     """
 
-    class AuthMethod(Enum):
-        """
-        Authentication method for OpenSearch
-        """
-
-        BASIC = "basic"
-        AWS_MANAGED_IAM = "aws_managed_iam"
-
     OPENSEARCH_HOST: str | None = Field(
         description="Hostname or IP address of the OpenSearch server (e.g., 'localhost' or 'opensearch.example.com')",
         default=None,

api/contexts/__init__.py

@@ -9,6 +9,7 @@ if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
+    from core.trigger.provider import PluginTriggerProviderController
 
 
 """
@@ -41,3 +42,11 @@ datasource_plugin_providers: RecyclableContextVar[dict[str, "DatasourcePluginPro
 datasource_plugin_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
     ContextVar("datasource_plugin_providers_lock")
 )
+
+plugin_trigger_providers: RecyclableContextVar[dict[str, "PluginTriggerProviderController"]] = RecyclableContextVar(
+    ContextVar("plugin_trigger_providers")
+)
+
+plugin_trigger_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
+    ContextVar("plugin_trigger_providers_lock")
+)

api/controllers/console/__init__.py

@@ -66,6 +66,7 @@ from .app import (
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
+    workflow_trigger,
 )
 
 # Import auth controllers
@@ -126,6 +127,7 @@ from .workspace import (
     models,
     plugin,
     tool_providers,
+    trigger_providers,
     workspace,
 )
 
@@ -196,6 +198,7 @@ __all__ = [
     "statistic",
     "tags",
     "tool_providers",
+    "trigger_providers",
     "version",
     "website",
     "workflow",
@@ -203,5 +206,6 @@ __all__ = [
     "workflow_draft_variable",
     "workflow_run",
     "workflow_statistic",
+    "workflow_trigger",
     "workspace",
 ]

api/controllers/console/apikey.py

@@ -1,5 +1,4 @@
 import flask_restx
-from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with
 from flask_restx._http import HTTPStatus
 from sqlalchemy import select
@@ -8,7 +7,8 @@ from werkzeug.exceptions import Forbidden
 
 from extensions.ext_database import db
 from libs.helper import TimestampField
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.account import Account
 from models.dataset import Dataset
 from models.model import ApiToken, App
 
@@ -57,6 +57,8 @@ class BaseApiKeyListResource(Resource):
     def get(self, resource_id):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
         keys = db.session.scalars(
             select(ApiToken).where(
@@ -69,8 +71,10 @@ class BaseApiKeyListResource(Resource):
     def post(self, resource_id):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
-        if not current_user.is_editor:
+        if not current_user.has_edit_permission:
             raise Forbidden()
 
         current_key_count = (
@@ -108,6 +112,8 @@ class BaseApiKeyResource(Resource):
         assert self.resource_id_field is not None, "resource_id_field must be set"
         resource_id = str(resource_id)
         api_key_id = str(api_key_id)
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         _get_resource(resource_id, current_user.current_tenant_id, self.resource_model)
 
         # The role of the current user in the ta table must be admin or owner

api/controllers/console/app/app.py

@@ -304,7 +304,7 @@ class AppCopyApi(Resource):
             account = cast(Account, current_user)
             result = import_service.import_app(
                 account=account,
-                import_mode=ImportMode.YAML_CONTENT.value,
+                import_mode=ImportMode.YAML_CONTENT,
                 yaml_content=yaml_content,
                 name=args.get("name"),
                 description=args.get("description"),

api/controllers/console/app/app_import.py

@@ -70,9 +70,9 @@ class AppImportApi(Resource):
             EnterpriseService.WebAppAuth.update_app_access_mode(result.app_id, "private")
         # Return appropriate status code based on result
         status = result.status
-        if status == ImportStatus.FAILED.value:
+        if status == ImportStatus.FAILED:
             return result.model_dump(mode="json"), 400
-        elif status == ImportStatus.PENDING.value:
+        elif status == ImportStatus.PENDING:
             return result.model_dump(mode="json"), 202
         return result.model_dump(mode="json"), 200
 
@@ -97,7 +97,7 @@ class AppImportConfirmApi(Resource):
             session.commit()
 
         # Return appropriate status code based on result
-        if result.status == ImportStatus.FAILED.value:
+        if result.status == ImportStatus.FAILED:
             return result.model_dump(mode="json"), 400
         return result.model_dump(mode="json"), 200
 

api/controllers/console/app/conversation.py

@@ -309,7 +309,7 @@ class ChatConversationApi(Resource):
             )
 
         if app_model.mode == AppMode.ADVANCED_CHAT:
-            query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER.value)
+            query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)
 
         match args["sort_by"]:
             case "created_at":

api/controllers/console/app/generator.py

@@ -12,6 +12,7 @@ from controllers.console.app.error import (
 )
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from core.helper.code_executor.code_node_provider import CodeNodeProvider
 from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
 from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.llm_generator import LLMGenerator
@@ -198,13 +199,11 @@ class InstructionGenerateApi(Resource):
         parser.add_argument("model_config", type=dict, required=True, nullable=False, location="json")
         parser.add_argument("ideal_output", type=str, required=False, default="", location="json")
         args = parser.parse_args()
-        code_template = (
-            Python3CodeProvider.get_default_code()
-            if args["language"] == "python"
-            else (JavascriptCodeProvider.get_default_code())
-            if args["language"] == "javascript"
-            else ""
+        providers: list[type[CodeNodeProvider]] = [Python3CodeProvider, JavascriptCodeProvider]
+        code_provider: type[CodeNodeProvider] | None = next(
+            (p for p in providers if p.is_accept_language(args["language"])), None
         )
+        code_template = code_provider.get_default_code() if code_provider else ""
         try:
             # Generate from nothing for a workflow node
             if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":

api/controllers/console/app/model_config.py

@@ -14,6 +14,7 @@ from core.tools.tool_manager import ToolManager
 from core.tools.utils.configuration import ToolParameterConfigurationManager
 from events.app_event import app_model_config_was_updated
 from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models.account import Account
 from models.model import AppMode, AppModelConfig
@@ -172,6 +173,8 @@ class ModelConfigResource(Resource):
         db.session.flush()
 
         app_model.app_model_config_id = new_app_model_config.id
+        app_model.updated_by = current_user.id
+        app_model.updated_at = naive_utc_now()
         db.session.commit()
 
         app_model_config_was_updated.send(app_model, app_model_config=new_app_model_config)

api/controllers/console/app/statistic.py

@@ -52,7 +52,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -127,7 +127,7 @@ class DailyConversationStatistic(Resource):
                 sa.func.count(sa.distinct(Message.conversation_id)).label("conversation_count"),
             )
             .select_from(Message)
-            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER.value)
+            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER)
         )
 
         if args["start"]:
@@ -190,7 +190,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -263,7 +263,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -345,7 +345,7 @@ FROM
         WHERE
             c.app_id = :app_id
             AND m.invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -432,7 +432,7 @@ LEFT JOIN
 WHERE
     m.app_id = :app_id
     AND m.invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -509,7 +509,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc
@@ -584,7 +584,7 @@ FROM
 WHERE
     app_id = :app_id
     AND invoke_from != :invoke_from"""
-        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER.value}
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
 
         timezone = pytz.timezone(account.timezone)
         utc_timezone = pytz.utc

api/controllers/console/app/workflow.py

@@ -19,21 +19,30 @@ from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
+from core.model_runtime.utils.encoders import jsonable_encoder
 from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from fields.workflow_run_fields import workflow_run_node_execution_fields
 from libs import helper
+from libs.datetime_utils import naive_utc_now
 from libs.helper import TimestampField, uuid_value
 from libs.login import current_user, login_required
 from models import App
 from models.account import Account
 from models.model import AppMode
-from models.workflow import Workflow
+from models.provider_ids import TriggerProviderID
+from models.workflow import NodeType, Workflow
 from services.app_generate_service import AppGenerateService
 from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
+from services.trigger.trigger_debug_service import (
+    PluginTriggerDebugEvent,
+    TriggerDebugService,
+    WebhookDebugEvent,
+)
+from services.trigger.webhook_service import WebhookService
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
 
 logger = logging.getLogger(__name__)
@@ -674,8 +683,12 @@ class PublishedWorkflowApi(Resource):
                 marked_comment=args.marked_comment or "",
             )
 
-            app_model.workflow_id = workflow.id
-            db.session.commit()  # NOTE: this is necessary for update app_model.workflow_id
+            # Update app_model within the same session to ensure atomicity
+            app_model_in_session = session.get(App, app_model.id)
+            if app_model_in_session:
+                app_model_in_session.workflow_id = workflow.id
+                app_model_in_session.updated_by = current_user.id
+                app_model_in_session.updated_at = naive_utc_now()
 
             workflow_created_at = TimestampField().format(workflow.created_at)
 
@@ -985,3 +998,417 @@ class DraftWorkflowNodeLastRunApi(Resource):
         if node_exec is None:
             raise NotFound("last run not found")
         return node_exec
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger")
+class DraftWorkflowTriggerNodeApi(Resource):
+    """
+    Single node debug - Polling API for trigger events
+    Path: /apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/trigger
+    """
+
+    @api.doc("poll_draft_workflow_trigger_node")
+    @api.doc(description="Poll for trigger events and execute single node when event arrives")
+    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @api.expect(
+        api.model(
+            "DraftWorkflowTriggerNodeRequest",
+            {
+                "event_name": fields.String(required=True, description="Event name"),
+                "subscription_id": fields.String(required=True, description="Subscription ID"),
+                "provider_id": fields.String(required=True, description="Provider ID"),
+            },
+        )
+    )
+    @api.response(200, "Trigger event received and node executed successfully")
+    @api.response(403, "Permission denied")
+    @api.response(500, "Internal server error")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    def post(self, app_model: App, node_id: str):
+        """
+        Poll for trigger events and execute single node when event arrives
+        """
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("event_name", type=str, required=True, location="json", nullable=False)
+        parser.add_argument("subscription_id", type=str, required=True, location="json", nullable=False)
+        parser.add_argument("provider_id", type=str, required=True, location="json", nullable=False)
+        args = parser.parse_args()
+        event_name = args["event_name"]
+        subscription_id = args["subscription_id"]
+        provider_id = args["provider_id"]
+
+        pool_key = PluginTriggerDebugEvent.build_pool_key(
+            tenant_id=app_model.tenant_id,
+            provider_id=provider_id,
+            subscription_id=subscription_id,
+            event_name=event_name,
+        )
+        event: PluginTriggerDebugEvent | None = TriggerDebugService.poll(
+            event_type=PluginTriggerDebugEvent,
+            pool_key=pool_key,
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            app_id=app_model.id,
+            node_id=node_id,
+        )
+        if not event:
+            return jsonable_encoder({"status": "waiting"})
+
+        try:
+            workflow_service = WorkflowService()
+            draft_workflow = workflow_service.get_draft_workflow(app_model)
+            if not draft_workflow:
+                raise ValueError("Workflow not found")
+
+            user_inputs = event.model_dump()
+            node_execution = workflow_service.run_draft_workflow_node(
+                app_model=app_model,
+                draft_workflow=draft_workflow,
+                node_id=node_id,
+                user_inputs=user_inputs,
+                account=current_user,
+                query="",
+                files=[],
+            )
+            return jsonable_encoder(node_execution)
+        except Exception:
+            logger.exception("Error running draft workflow trigger node")
+            return jsonable_encoder(
+                {
+                    "status": "error",
+                }
+            ), 500
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/plugin/run")
+class DraftWorkflowTriggerRunApi(Resource):
+    """
+    Full workflow debug - Polling API for trigger events
+    Path: /apps/<uuid:app_id>/workflows/draft/trigger/run
+    """
+
+    @api.doc("poll_draft_workflow_trigger_run")
+    @api.doc(description="Poll for trigger events and execute full workflow when event arrives")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.expect(
+        api.model(
+            "DraftWorkflowTriggerRunRequest",
+            {
+                "node_id": fields.String(required=True, description="Node ID"),
+            },
+        )
+    )
+    @api.response(200, "Trigger event received and workflow executed successfully")
+    @api.response(403, "Permission denied")
+    @api.response(500, "Internal server error")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        """
+        Poll for trigger events and execute full workflow when event arrives
+        """
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
+        args = parser.parse_args()
+        node_id = args["node_id"]
+        workflow_service = WorkflowService()
+        workflow: Workflow | None = workflow_service.get_draft_workflow(
+            app_model=app_model,
+            workflow_id=None,
+        )
+
+        if not workflow:
+            return jsonable_encoder({"status": "error", "message": "Workflow not found"}), 404
+
+        node_data = workflow.get_node_config_by_id(node_id=node_id).get("data")
+        if not node_data:
+            return jsonable_encoder({"status": "error", "message": "Node config not found"}), 404
+
+        event_name = node_data.get("event_name")
+        subscription_id = node_data.get("subscription_id")
+        if not subscription_id:
+            return jsonable_encoder({"status": "error", "message": "Subscription ID not found"}), 404
+
+        provider_id = TriggerProviderID(node_data.get("provider_id"))
+        pool_key: str = PluginTriggerDebugEvent.build_pool_key(
+            tenant_id=app_model.tenant_id,
+            provider_id=provider_id,
+            subscription_id=subscription_id,
+            event_name=event_name,
+        )
+        event: PluginTriggerDebugEvent | None = TriggerDebugService.poll(
+            event_type=PluginTriggerDebugEvent,
+            pool_key=pool_key,
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            app_id=app_model.id,
+            node_id=node_id,
+        )
+        if not event:
+            return jsonable_encoder({"status": "waiting", "retry_in": 2000})
+
+        workflow_args = {
+            "inputs": event.model_dump(),
+            "query": "",
+            "files": [],
+        }
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            workflow_args["external_trace_id"] = external_trace_id
+
+        try:
+            response = AppGenerateService.generate(
+                app_model=app_model,
+                user=current_user,
+                args=workflow_args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+                root_node_id=node_id,
+            )
+            return helper.compact_generate_response(response)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except Exception:
+            logger.exception("Error running draft workflow trigger run")
+            return jsonable_encoder(
+                {
+                    "status": "error",
+                }
+            ), 500
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/webhook/run")
+class DraftWorkflowTriggerWebhookRunApi(Resource):
+    """
+    Full workflow debug when the start node is a webhook trigger
+    Path: /apps/<uuid:app_id>/workflows/draft/trigger/webhook/run
+    """
+
+    @api.doc("draft_workflow_trigger_webhook_run")
+    @api.doc(description="Full workflow debug when the start node is a webhook trigger")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.expect(
+        api.model(
+            "DraftWorkflowTriggerWebhookRunRequest",
+            {
+                "node_id": fields.String(required=True, description="Node ID"),
+            },
+        )
+    )
+    @api.response(200, "Workflow executed successfully")
+    @api.response(403, "Permission denied")
+    @api.response(500, "Internal server error")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        """
+        Full workflow debug when the start node is a webhook trigger
+        """
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
+        args = parser.parse_args()
+        node_id = args["node_id"]
+
+        pool_key = WebhookDebugEvent.build_pool_key(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            node_id=node_id,
+        )
+        event: WebhookDebugEvent | None = TriggerDebugService.poll(
+            event_type=WebhookDebugEvent,
+            pool_key=pool_key,
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            app_id=app_model.id,
+            node_id=node_id,
+        )
+
+        if not event:
+            return jsonable_encoder({"status": "waiting", "retry_in": 2000})
+
+        payload = event.payload or {}
+        workflow_inputs = payload.get("inputs")
+        if workflow_inputs is None:
+            webhook_data = payload.get("webhook_data", {})
+            workflow_inputs = WebhookService.build_workflow_inputs(webhook_data)
+
+        workflow_args = {
+            "inputs": workflow_inputs or {},
+            "query": "",
+            "files": [],
+        }
+
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            workflow_args["external_trace_id"] = external_trace_id
+
+        try:
+            response = AppGenerateService.generate(
+                app_model=app_model,
+                user=current_user,
+                args=workflow_args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+                root_node_id=node_id,
+            )
+            return helper.compact_generate_response(response)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except Exception:
+            logger.exception("Error running draft workflow trigger webhook run")
+            return jsonable_encoder(
+                {
+                    "status": "error",
+                }
+            ), 500
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/debug/webhook/run")
+class DraftWorkflowNodeWebhookDebugRunApi(Resource):
+    """Single node debug when the node is a webhook trigger."""
+
+    @api.doc("draft_workflow_node_webhook_debug_run")
+    @api.doc(description="Poll for webhook debug payload and execute single node when event arrives")
+    @api.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
+    @api.response(200, "Node executed successfully")
+    @api.response(403, "Permission denied")
+    @api.response(400, "Invalid node type")
+    @api.response(500, "Internal server error")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    def post(self, app_model: App, node_id: str):
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        pool_key = WebhookDebugEvent.build_pool_key(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            node_id=node_id,
+        )
+        event: WebhookDebugEvent | None = TriggerDebugService.poll(
+            event_type=WebhookDebugEvent,
+            pool_key=pool_key,
+            tenant_id=app_model.tenant_id,
+            user_id=current_user.id,
+            app_id=app_model.id,
+            node_id=node_id,
+        )
+
+        if not event:
+            return jsonable_encoder({"status": "waiting", "retry_in": 2000})
+
+        workflow_service = WorkflowService()
+        draft_workflow = workflow_service.get_draft_workflow(app_model=app_model)
+
+        if not draft_workflow:
+            raise DraftWorkflowNotExist()
+
+        node_config = draft_workflow.get_node_config_by_id(node_id)
+        node_type = Workflow.get_node_type_from_node_config(node_config)
+        if node_type != NodeType.TRIGGER_WEBHOOK:
+            return jsonable_encoder(
+                {
+                    "status": "error",
+                    "message": "node is not webhook trigger",
+                }
+            ), 400
+
+        payload = event.payload or {}
+        workflow_inputs = payload.get("inputs")
+        if workflow_inputs is None:
+            webhook_data = payload.get("webhook_data", {})
+            workflow_inputs = WebhookService.build_workflow_inputs(webhook_data)
+
+        workflow_node_execution = workflow_service.run_draft_workflow_node(
+            app_model=app_model,
+            draft_workflow=draft_workflow,
+            node_id=node_id,
+            user_inputs=workflow_inputs or {},
+            account=current_user,
+            query="",
+            files=[],
+        )
+
+        return jsonable_encoder(workflow_node_execution)
+
+
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/trigger/schedule/run")
+class DraftWorkflowTriggerScheduleRunApi(Resource):
+    """
+    Full workflow debug when the start node is a schedule trigger
+    Path: /apps/<uuid:app_id>/workflows/draft/trigger/schedule/run
+    """
+
+    @api.doc("draft_workflow_trigger_schedule_run")
+    @api.doc(description="Full workflow debug when the start node is a schedule trigger")
+    @api.doc(params={"app_id": "Application ID"})
+    @api.expect(
+        api.model(
+            "DraftWorkflowTriggerScheduleRunRequest",
+            {
+                "node_id": fields.String(required=True, description="Node ID"),
+            },
+        )
+    )
+    @api.response(200, "Workflow executed successfully")
+    @api.response(403, "Permission denied")
+    @api.response(500, "Internal server error")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.WORKFLOW])
+    def post(self, app_model: App):
+        """
+        Full workflow debug when the start node is a schedule trigger
+        """
+        if not isinstance(current_user, Account) or not current_user.has_edit_permission:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, location="json", nullable=False)
+        args = parser.parse_args()
+        node_id = args["node_id"]
+
+        workflow_args = {
+            "inputs": {},
+            "query": "",
+            "files": [],
+        }
+
+        try:
+            response = AppGenerateService.generate(
+                app_model=app_model,
+                user=current_user,
+                args=workflow_args,
+                invoke_from=InvokeFrom.DEBUGGER,
+                streaming=True,
+                root_node_id=node_id,
+            )
+            return helper.compact_generate_response(response)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except Exception:
+            logger.exception("Error running draft workflow trigger schedule run")
+            return jsonable_encoder(
+                {
+                    "status": "error",
+                }
+            ), 500

api/controllers/console/app/workflow_statistic.py

@@ -47,7 +47,7 @@ WHERE
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
         }
 
         timezone = pytz.timezone(account.timezone)
@@ -115,7 +115,7 @@ WHERE
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
         }
 
         timezone = pytz.timezone(account.timezone)
@@ -183,7 +183,7 @@ WHERE
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
         }
 
         timezone = pytz.timezone(account.timezone)
@@ -269,7 +269,7 @@ GROUP BY
         arg_dict = {
             "tz": account.timezone,
             "app_id": app_model.id,
-            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN.value,
+            "triggered_from": WorkflowRunTriggeredFrom.APP_RUN,
         }
 
         timezone = pytz.timezone(account.timezone)

api/controllers/console/app/workflow_trigger.py

@@ -0,0 +1,149 @@
+import logging
+
+from flask_restx import Resource, marshal_with, reqparse
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden, NotFound
+
+from configs import dify_config
+from controllers.console import api
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, setup_required
+from extensions.ext_database import db
+from fields.workflow_trigger_fields import trigger_fields, triggers_list_fields, webhook_trigger_fields
+from libs.login import current_user, login_required
+from models.model import Account, AppMode
+from models.workflow import AppTrigger, AppTriggerStatus, WorkflowWebhookTrigger
+
+logger = logging.getLogger(__name__)
+
+
+class WebhookTriggerApi(Resource):
+    """Webhook Trigger API"""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    @marshal_with(webhook_trigger_fields)
+    def get(self, app_model):
+        """Get webhook trigger for a node"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("node_id", type=str, required=True, help="Node ID is required")
+        args = parser.parse_args()
+
+        node_id = args["node_id"]
+
+        with Session(db.engine) as session:
+            # Get webhook trigger for this app and node
+            webhook_trigger = (
+                session.query(WorkflowWebhookTrigger)
+                .filter(
+                    WorkflowWebhookTrigger.app_id == app_model.id,
+                    WorkflowWebhookTrigger.node_id == node_id,
+                )
+                .first()
+            )
+
+            if not webhook_trigger:
+                raise NotFound("Webhook trigger not found for this node")
+
+            # Add computed fields for marshal_with
+            base_url = dify_config.SERVICE_API_URL
+            webhook_trigger.webhook_url = f"{base_url}/triggers/webhook/{webhook_trigger.webhook_id}"  # type: ignore
+            webhook_trigger.webhook_debug_url = f"{base_url}/triggers/webhook-debug/{webhook_trigger.webhook_id}"  # type: ignore
+
+            return webhook_trigger
+
+
+class AppTriggersApi(Resource):
+    """App Triggers list API"""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    @marshal_with(triggers_list_fields)
+    def get(self, app_model):
+        """Get app triggers list"""
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
+
+        with Session(db.engine) as session:
+            # Get all triggers for this app using select API
+            triggers = (
+                session.execute(
+                    select(AppTrigger)
+                    .where(
+                        AppTrigger.tenant_id == current_user.current_tenant_id,
+                        AppTrigger.app_id == app_model.id,
+                    )
+                    .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc())
+                )
+                .scalars()
+                .all()
+            )
+
+        # Add computed icon field for each trigger
+        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
+        for trigger in triggers:
+            if trigger.trigger_type == "trigger-plugin":
+                trigger.icon = url_prefix + trigger.provider_name + "/icon"  # type: ignore
+            else:
+                trigger.icon = ""  # type: ignore
+
+        return {"data": triggers}
+
+
+class AppTriggerEnableApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=AppMode.WORKFLOW)
+    @marshal_with(trigger_fields)
+    def post(self, app_model):
+        """Update app trigger (enable/disable)"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("trigger_id", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("enable_trigger", type=bool, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
+        if not current_user.has_edit_permission:
+            raise Forbidden()
+
+        trigger_id = args["trigger_id"]
+
+        with Session(db.engine) as session:
+            # Find the trigger using select
+            trigger = session.execute(
+                select(AppTrigger).where(
+                    AppTrigger.id == trigger_id,
+                    AppTrigger.tenant_id == current_user.current_tenant_id,
+                    AppTrigger.app_id == app_model.id,
+                )
+            ).scalar_one_or_none()
+
+            if not trigger:
+                raise NotFound("Trigger not found")
+
+            # Update status based on enable_trigger boolean
+            trigger.status = AppTriggerStatus.ENABLED if args["enable_trigger"] else AppTriggerStatus.DISABLED
+
+            session.commit()
+            session.refresh(trigger)
+
+        # Add computed icon field
+        url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/"
+        if trigger.trigger_type == "trigger-plugin":
+            trigger.icon = url_prefix + trigger.provider_name + "/icon"  # type: ignore
+        else:
+            trigger.icon = ""  # type: ignore
+
+        return trigger
+
+
+api.add_resource(WebhookTriggerApi, "/apps/<uuid:app_id>/workflows/triggers/webhook")
+api.add_resource(AppTriggersApi, "/apps/<uuid:app_id>/triggers")
+api.add_resource(AppTriggerEnableApi, "/apps/<uuid:app_id>/trigger-enable")

api/controllers/console/auth/activate.py

@@ -103,7 +103,7 @@ class ActivateApi(Resource):
         account.interface_language = args["interface_language"]
         account.timezone = args["timezone"]
         account.interface_theme = "light"
-        account.status = AccountStatus.ACTIVE.value
+        account.status = AccountStatus.ACTIVE
         account.initialized_at = naive_utc_now()
         db.session.commit()
 

api/controllers/console/auth/oauth.py

@@ -130,11 +130,11 @@ class OAuthCallback(Resource):
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message={e.description}")
 
         # Check account status
-        if account.status == AccountStatus.BANNED.value:
+        if account.status == AccountStatus.BANNED:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Account is banned.")
 
-        if account.status == AccountStatus.PENDING.value:
-            account.status = AccountStatus.ACTIVE.value
+        if account.status == AccountStatus.PENDING:
+            account.status = AccountStatus.ACTIVE
             account.initialized_at = naive_utc_now()
             db.session.commit()
 

api/controllers/console/billing/compliance.py

@@ -1,9 +1,9 @@
 from flask import request
-from flask_login import current_user
 from flask_restx import Resource, reqparse
 
 from libs.helper import extract_remote_ip
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.account import Account
 from services.billing_service import BillingService
 
 from .. import console_ns
@@ -17,6 +17,8 @@ class ComplianceApi(Resource):
     @account_initialization_required
     @only_edition_cloud
     def get(self):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         parser = reqparse.RequestParser()
         parser.add_argument("doc_name", type=str, required=True, location="args")
         args = parser.parse_args()

api/controllers/console/datasets/data_source.py

@@ -256,7 +256,7 @@ class DataSourceNotionApi(Resource):
             credential_id = notion_info.get("credential_id")
             for page in notion_info["pages"]:
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION.value,
+                    datasource_type=DatasourceType.NOTION,
                     notion_info=NotionInfo.model_validate(
                         {
                             "credential_id": credential_id,

api/controllers/console/datasets/datasets.py

@@ -45,6 +45,79 @@ def _validate_name(name: str) -> str:
     return name
 
 
+def _get_retrieval_methods_by_vector_type(vector_type: str | None, is_mock: bool = False) -> dict[str, list[str]]:
+    """
+    Get supported retrieval methods based on vector database type.
+
+    Args:
+        vector_type: Vector database type, can be None
+        is_mock: Whether this is a Mock API, affects MILVUS handling
+
+    Returns:
+        Dictionary containing supported retrieval methods
+
+    Raises:
+        ValueError: If vector_type is None or unsupported
+    """
+    if vector_type is None:
+        raise ValueError("Vector store type is not configured.")
+
+    # Define vector database types that only support semantic search
+    semantic_only_types = {
+        VectorType.RELYT,
+        VectorType.TIDB_VECTOR,
+        VectorType.CHROMA,
+        VectorType.PGVECTO_RS,
+        VectorType.VIKINGDB,
+        VectorType.UPSTASH,
+    }
+
+    # Define vector database types that support all retrieval methods
+    full_search_types = {
+        VectorType.QDRANT,
+        VectorType.WEAVIATE,
+        VectorType.OPENSEARCH,
+        VectorType.ANALYTICDB,
+        VectorType.MYSCALE,
+        VectorType.ORACLE,
+        VectorType.ELASTICSEARCH,
+        VectorType.ELASTICSEARCH_JA,
+        VectorType.PGVECTOR,
+        VectorType.VASTBASE,
+        VectorType.TIDB_ON_QDRANT,
+        VectorType.LINDORM,
+        VectorType.COUCHBASE,
+        VectorType.OPENGAUSS,
+        VectorType.OCEANBASE,
+        VectorType.TABLESTORE,
+        VectorType.HUAWEI_CLOUD,
+        VectorType.TENCENT,
+        VectorType.MATRIXONE,
+        VectorType.CLICKZETTA,
+        VectorType.BAIDU,
+        VectorType.ALIBABACLOUD_MYSQL,
+    }
+
+    semantic_methods = {"retrieval_method": [RetrievalMethod.SEMANTIC_SEARCH.value]}
+    full_methods = {
+        "retrieval_method": [
+            RetrievalMethod.SEMANTIC_SEARCH.value,
+            RetrievalMethod.FULL_TEXT_SEARCH.value,
+            RetrievalMethod.HYBRID_SEARCH.value,
+        ]
+    }
+
+    if vector_type == VectorType.MILVUS:
+        return semantic_methods if is_mock else full_methods
+
+    if vector_type in semantic_only_types:
+        return semantic_methods
+    elif vector_type in full_search_types:
+        return full_methods
+    else:
+        raise ValueError(f"Unsupported vector db type {vector_type}.")
+
+
 @console_ns.route("/datasets")
 class DatasetListApi(Resource):
     @api.doc("get_datasets")
@@ -500,7 +573,7 @@ class DatasetIndexingEstimateApi(Resource):
             if file_details:
                 for file_detail in file_details:
                     extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.FILE.value,
+                        datasource_type=DatasourceType.FILE,
                         upload_file=file_detail,
                         document_model=args["doc_form"],
                     )
@@ -512,7 +585,7 @@ class DatasetIndexingEstimateApi(Resource):
                 credential_id = notion_info.get("credential_id")
                 for page in notion_info["pages"]:
                     extract_setting = ExtractSetting(
-                        datasource_type=DatasourceType.NOTION.value,
+                        datasource_type=DatasourceType.NOTION,
                         notion_info=NotionInfo.model_validate(
                             {
                                 "credential_id": credential_id,
@@ -529,7 +602,7 @@ class DatasetIndexingEstimateApi(Resource):
             website_info_list = args["info_list"]["website_info_list"]
             for url in website_info_list["urls"]:
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE.value,
+                    datasource_type=DatasourceType.WEBSITE,
                     website_info=WebsiteInfo.model_validate(
                         {
                             "provider": website_info_list["provider"],
@@ -777,49 +850,7 @@ class DatasetRetrievalSettingApi(Resource):
     @account_initialization_required
     def get(self):
         vector_type = dify_config.VECTOR_STORE
-        match vector_type:
-            case (
-                VectorType.RELYT
-                | VectorType.TIDB_VECTOR
-                | VectorType.CHROMA
-                | VectorType.PGVECTO_RS
-                | VectorType.VIKINGDB
-                | VectorType.UPSTASH
-            ):
-                return {"retrieval_method": [RetrievalMethod.SEMANTIC_SEARCH.value]}
-            case (
-                VectorType.QDRANT
-                | VectorType.WEAVIATE
-                | VectorType.OPENSEARCH
-                | VectorType.ANALYTICDB
-                | VectorType.MYSCALE
-                | VectorType.ORACLE
-                | VectorType.ELASTICSEARCH
-                | VectorType.ELASTICSEARCH_JA
-                | VectorType.PGVECTOR
-                | VectorType.VASTBASE
-                | VectorType.TIDB_ON_QDRANT
-                | VectorType.LINDORM
-                | VectorType.COUCHBASE
-                | VectorType.MILVUS
-                | VectorType.OPENGAUSS
-                | VectorType.OCEANBASE
-                | VectorType.TABLESTORE
-                | VectorType.HUAWEI_CLOUD
-                | VectorType.TENCENT
-                | VectorType.MATRIXONE
-                | VectorType.CLICKZETTA
-                | VectorType.BAIDU
-            ):
-                return {
-                    "retrieval_method": [
-                        RetrievalMethod.SEMANTIC_SEARCH.value,
-                        RetrievalMethod.FULL_TEXT_SEARCH.value,
-                        RetrievalMethod.HYBRID_SEARCH.value,
-                    ]
-                }
-            case _:
-                raise ValueError(f"Unsupported vector db type {vector_type}.")
+        return _get_retrieval_methods_by_vector_type(vector_type, is_mock=False)
 
 
 @console_ns.route("/datasets/retrieval-setting/<string:vector_type>")
@@ -832,48 +863,7 @@ class DatasetRetrievalSettingMockApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, vector_type):
-        match vector_type:
-            case (
-                VectorType.MILVUS
-                | VectorType.RELYT
-                | VectorType.TIDB_VECTOR
-                | VectorType.CHROMA
-                | VectorType.PGVECTO_RS
-                | VectorType.VIKINGDB
-                | VectorType.UPSTASH
-            ):
-                return {"retrieval_method": [RetrievalMethod.SEMANTIC_SEARCH.value]}
-            case (
-                VectorType.QDRANT
-                | VectorType.WEAVIATE
-                | VectorType.OPENSEARCH
-                | VectorType.ANALYTICDB
-                | VectorType.MYSCALE
-                | VectorType.ORACLE
-                | VectorType.ELASTICSEARCH
-                | VectorType.ELASTICSEARCH_JA
-                | VectorType.COUCHBASE
-                | VectorType.PGVECTOR
-                | VectorType.VASTBASE
-                | VectorType.LINDORM
-                | VectorType.OPENGAUSS
-                | VectorType.OCEANBASE
-                | VectorType.TABLESTORE
-                | VectorType.TENCENT
-                | VectorType.HUAWEI_CLOUD
-                | VectorType.MATRIXONE
-                | VectorType.CLICKZETTA
-                | VectorType.BAIDU
-            ):
-                return {
-                    "retrieval_method": [
-                        RetrievalMethod.SEMANTIC_SEARCH.value,
-                        RetrievalMethod.FULL_TEXT_SEARCH.value,
-                        RetrievalMethod.HYBRID_SEARCH.value,
-                    ]
-                }
-            case _:
-                raise ValueError(f"Unsupported vector db type {vector_type}.")
+        return _get_retrieval_methods_by_vector_type(vector_type, is_mock=True)
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/error-docs")

api/controllers/console/datasets/datasets_document.py

@@ -475,7 +475,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE.value, upload_file=file, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE, upload_file=file, document_model=document.doc_form
                 )
 
                 indexing_runner = IndexingRunner()
@@ -538,7 +538,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE.value, upload_file=file_detail, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE, upload_file=file_detail, document_model=document.doc_form
                 )
                 extract_settings.append(extract_setting)
 
@@ -546,7 +546,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 if not data_source_info:
                     continue
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.NOTION.value,
+                    datasource_type=DatasourceType.NOTION,
                     notion_info=NotionInfo.model_validate(
                         {
                             "credential_id": data_source_info["credential_id"],
@@ -563,7 +563,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 if not data_source_info:
                     continue
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.WEBSITE.value,
+                    datasource_type=DatasourceType.WEBSITE,
                     website_info=WebsiteInfo.model_validate(
                         {
                             "provider": data_source_info["provider"],

api/controllers/console/datasets/hit_testing_base.py

@@ -1,7 +1,5 @@
 import logging
-from typing import cast
 
-from flask_login import current_user
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
@@ -21,6 +19,7 @@ from core.errors.error import (
 )
 from core.model_runtime.errors.invoke import InvokeError
 from fields.hit_testing_fields import hit_testing_record_fields
+from libs.login import current_user
 from models.account import Account
 from services.dataset_service import DatasetService
 from services.hit_testing_service import HitTestingService
@@ -31,6 +30,7 @@ logger = logging.getLogger(__name__)
 class DatasetsHitTestingBase:
     @staticmethod
     def get_and_validate_dataset(dataset_id: str):
+        assert isinstance(current_user, Account)
         dataset = DatasetService.get_dataset(dataset_id)
         if dataset is None:
             raise NotFound("Dataset not found.")
@@ -57,11 +57,12 @@ class DatasetsHitTestingBase:
 
     @staticmethod
     def perform_hit_testing(dataset, args):
+        assert isinstance(current_user, Account)
         try:
             response = HitTestingService.retrieve(
                 dataset=dataset,
                 query=args["query"],
-                account=cast(Account, current_user),
+                account=current_user,
                 retrieval_model=args["retrieval_model"],
                 external_retrieval_model=args["external_retrieval_model"],
                 limit=10,

api/controllers/console/datasets/rag_pipeline/rag_pipeline_import.py

@@ -60,9 +60,9 @@ class RagPipelineImportApi(Resource):
 
         # Return appropriate status code based on result
         status = result.status
-        if status == ImportStatus.FAILED.value:
+        if status == ImportStatus.FAILED:
             return result.model_dump(mode="json"), 400
-        elif status == ImportStatus.PENDING.value:
+        elif status == ImportStatus.PENDING:
             return result.model_dump(mode="json"), 202
         return result.model_dump(mode="json"), 200
 
@@ -87,7 +87,7 @@ class RagPipelineImportConfirmApi(Resource):
             session.commit()
 
         # Return appropriate status code based on result
-        if result.status == ImportStatus.FAILED.value:
+        if result.status == ImportStatus.FAILED:
             return result.model_dump(mode="json"), 400
         return result.model_dump(mode="json"), 200
 

api/controllers/console/explore/wraps.py

@@ -2,15 +2,15 @@ from collections.abc import Callable
 from functools import wraps
 from typing import Concatenate, ParamSpec, TypeVar
 
-from flask_login import current_user
 from flask_restx import Resource
 from werkzeug.exceptions import NotFound
 
 from controllers.console.explore.error import AppAccessDeniedError
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
-from libs.login import login_required
+from libs.login import current_user, login_required
 from models import InstalledApp
+from models.account import Account
 from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
@@ -24,6 +24,8 @@ def installed_app_required(view: Callable[Concatenate[InstalledApp, P], R] | Non
     def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
         def decorated(installed_app_id: str, *args: P.args, **kwargs: P.kwargs):
+            assert isinstance(current_user, Account)
+            assert current_user.current_tenant_id is not None
             installed_app = (
                 db.session.query(InstalledApp)
                 .where(
@@ -56,6 +58,7 @@ def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] |
         def decorated(installed_app: InstalledApp, *args: P.args, **kwargs: P.kwargs):
             feature = FeatureService.get_system_features()
             if feature.webapp_auth.enabled:
+                assert isinstance(current_user, Account)
                 app_id = installed_app.app_id
                 app_code = AppService.get_app_code_by_id(app_id)
                 res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(

api/controllers/console/extension.py

@@ -1,11 +1,11 @@
-from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with, reqparse
 
 from constants import HIDDEN_VALUE
 from controllers.console import api, console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.api_based_extension_fields import api_based_extension_fields
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.account import Account
 from models.api_based_extension import APIBasedExtension
 from services.api_based_extension_service import APIBasedExtensionService
 from services.code_based_extension_service import CodeBasedExtensionService
@@ -47,6 +47,8 @@ class APIBasedExtensionAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def get(self):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         tenant_id = current_user.current_tenant_id
         return APIBasedExtensionService.get_all_by_tenant_id(tenant_id)
 
@@ -68,6 +70,8 @@ class APIBasedExtensionAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def post(self):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, location="json")
         parser.add_argument("api_endpoint", type=str, required=True, location="json")
@@ -95,6 +99,8 @@ class APIBasedExtensionDetailAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def get(self, id):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         api_based_extension_id = str(id)
         tenant_id = current_user.current_tenant_id
 
@@ -119,6 +125,8 @@ class APIBasedExtensionDetailAPI(Resource):
     @account_initialization_required
     @marshal_with(api_based_extension_fields)
     def post(self, id):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         api_based_extension_id = str(id)
         tenant_id = current_user.current_tenant_id
 
@@ -146,6 +154,8 @@ class APIBasedExtensionDetailAPI(Resource):
     @login_required
     @account_initialization_required
     def delete(self, id):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         api_based_extension_id = str(id)
         tenant_id = current_user.current_tenant_id
 

api/controllers/console/feature.py

@@ -1,7 +1,7 @@
-from flask_login import current_user
 from flask_restx import Resource, fields
 
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.account import Account
 from services.feature_service import FeatureService
 
 from . import api, console_ns
@@ -23,6 +23,8 @@ class FeatureApi(Resource):
     @cloud_utm_record
     def get(self):
         """Get feature configuration for current tenant"""
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         return FeatureService.get_features(current_user.current_tenant_id).model_dump()
 
 

api/controllers/console/remote_files.py

@@ -1,8 +1,6 @@
 import urllib.parse
-from typing import cast
 
 import httpx
-from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 
 import services
@@ -16,6 +14,7 @@ from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from extensions.ext_database import db
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
+from libs.login import current_user
 from models.account import Account
 from services.file_service import FileService
 
@@ -65,7 +64,8 @@ class RemoteFileUploadApi(Resource):
         content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
 
         try:
-            user = cast(Account, current_user)
+            assert isinstance(current_user, Account)
+            user = current_user
             upload_file = FileService(db.engine).upload_file(
                 filename=file_info.filename,
                 content=content,

api/controllers/console/tag/tags.py

@@ -1,12 +1,12 @@
 from flask import request
-from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from fields.tag_fields import dataset_tag_fields
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.account import Account
 from models.model import Tag
 from services.tag_service import TagService
 
@@ -24,6 +24,8 @@ class TagListApi(Resource):
     @account_initialization_required
     @marshal_with(dataset_tag_fields)
     def get(self):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         tag_type = request.args.get("type", type=str, default="")
         keyword = request.args.get("keyword", default=None, type=str)
         tags = TagService.get_tags(tag_type, current_user.current_tenant_id, keyword)
@@ -34,8 +36,10 @@ class TagListApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not (current_user.is_editor or current_user.is_dataset_editor):
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -59,9 +63,11 @@ class TagUpdateDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def patch(self, tag_id):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         tag_id = str(tag_id)
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not (current_user.is_editor or current_user.is_dataset_editor):
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -81,9 +87,11 @@ class TagUpdateDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def delete(self, tag_id):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         tag_id = str(tag_id)
         # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor:
+        if not current_user.has_edit_permission:
             raise Forbidden()
 
         TagService.delete_tag(tag_id)
@@ -97,8 +105,10 @@ class TagBindingCreateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
-        if not (current_user.is_editor or current_user.is_dataset_editor):
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()
@@ -123,8 +133,10 @@ class TagBindingDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
+        assert isinstance(current_user, Account)
+        assert current_user.current_tenant_id is not None
         # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
-        if not (current_user.is_editor or current_user.is_dataset_editor):
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
             raise Forbidden()
 
         parser = reqparse.RequestParser()

api/controllers/console/workspace/agent_providers.py

@@ -1,10 +1,10 @@
-from flask_login import current_user
 from flask_restx import Resource, fields
 
 from controllers.console import api, console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.account import Account
 from services.agent_service import AgentService
 
 
@@ -21,7 +21,9 @@ class AgentProviderListApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
+        assert isinstance(current_user, Account)
         user = current_user
+        assert user.current_tenant_id is not None
 
         user_id = user.id
         tenant_id = user.current_tenant_id
@@ -43,7 +45,9 @@ class AgentProviderApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, provider_name: str):
+        assert isinstance(current_user, Account)
         user = current_user
+        assert user.current_tenant_id is not None
         user_id = user.id
         tenant_id = user.current_tenant_id
         return jsonable_encoder(AgentService.get_agent_provider(user_id, tenant_id, provider_name))

api/controllers/console/workspace/endpoint.py

@@ -1,4 +1,3 @@
-from flask_login import current_user
 from flask_restx import Resource, fields, reqparse
 from werkzeug.exceptions import Forbidden
 
@@ -6,10 +5,18 @@ from controllers.console import api, console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.exc import PluginPermissionDeniedError
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.account import Account
 from services.plugin.endpoint_service import EndpointService
 
 
+def _current_account_with_tenant() -> tuple[Account, str]:
+    assert isinstance(current_user, Account)
+    tenant_id = current_user.current_tenant_id
+    assert tenant_id is not None
+    return current_user, tenant_id
+
+
 @console_ns.route("/workspaces/current/endpoints/create")
 class EndpointCreateApi(Resource):
     @api.doc("create_endpoint")
@@ -34,7 +41,7 @@ class EndpointCreateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user = current_user
+        user, tenant_id = _current_account_with_tenant()
         if not user.is_admin_or_owner:
             raise Forbidden()
 
@@ -51,7 +58,7 @@ class EndpointCreateApi(Resource):
         try:
             return {
                 "success": EndpointService.create_endpoint(
-                    tenant_id=user.current_tenant_id,
+                    tenant_id=tenant_id,
                     user_id=user.id,
                     plugin_unique_identifier=plugin_unique_identifier,
                     name=name,
@@ -80,7 +87,7 @@ class EndpointListApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        user = current_user
+        user, tenant_id = _current_account_with_tenant()
 
         parser = reqparse.RequestParser()
         parser.add_argument("page", type=int, required=True, location="args")
@@ -93,7 +100,7 @@ class EndpointListApi(Resource):
         return jsonable_encoder(
             {
                 "endpoints": EndpointService.list_endpoints(
-                    tenant_id=user.current_tenant_id,
+                    tenant_id=tenant_id,
                     user_id=user.id,
                     page=page,
                     page_size=page_size,
@@ -123,7 +130,7 @@ class EndpointListForSinglePluginApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        user = current_user
+        user, tenant_id = _current_account_with_tenant()
 
         parser = reqparse.RequestParser()
         parser.add_argument("page", type=int, required=True, location="args")
@@ -138,7 +145,7 @@ class EndpointListForSinglePluginApi(Resource):
         return jsonable_encoder(
             {
                 "endpoints": EndpointService.list_endpoints_for_single_plugin(
-                    tenant_id=user.current_tenant_id,
+                    tenant_id=tenant_id,
                     user_id=user.id,
                     plugin_id=plugin_id,
                     page=page,
@@ -165,7 +172,7 @@ class EndpointDeleteApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user = current_user
+        user, tenant_id = _current_account_with_tenant()
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -177,9 +184,7 @@ class EndpointDeleteApi(Resource):
         endpoint_id = args["endpoint_id"]
 
         return {
-            "success": EndpointService.delete_endpoint(
-                tenant_id=user.current_tenant_id, user_id=user.id, endpoint_id=endpoint_id
-            )
+            "success": EndpointService.delete_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
         }
 
 
@@ -207,7 +212,7 @@ class EndpointUpdateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user = current_user
+        user, tenant_id = _current_account_with_tenant()
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -224,7 +229,7 @@ class EndpointUpdateApi(Resource):
 
         return {
             "success": EndpointService.update_endpoint(
-                tenant_id=user.current_tenant_id,
+                tenant_id=tenant_id,
                 user_id=user.id,
                 endpoint_id=endpoint_id,
                 name=name,
@@ -250,7 +255,7 @@ class EndpointEnableApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user = current_user
+        user, tenant_id = _current_account_with_tenant()
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -262,9 +267,7 @@ class EndpointEnableApi(Resource):
             raise Forbidden()
 
         return {
-            "success": EndpointService.enable_endpoint(
-                tenant_id=user.current_tenant_id, user_id=user.id, endpoint_id=endpoint_id
-            )
+            "success": EndpointService.enable_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
         }
 
 
@@ -285,7 +288,7 @@ class EndpointDisableApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        user = current_user
+        user, tenant_id = _current_account_with_tenant()
 
         parser = reqparse.RequestParser()
         parser.add_argument("endpoint_id", type=str, required=True)
@@ -297,7 +300,5 @@ class EndpointDisableApi(Resource):
             raise Forbidden()
 
         return {
-            "success": EndpointService.disable_endpoint(
-                tenant_id=user.current_tenant_id, user_id=user.id, endpoint_id=endpoint_id
-            )
+            "success": EndpointService.disable_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
         }

api/controllers/console/workspace/members.py

@@ -1,7 +1,6 @@
 from urllib import parse
 
 from flask import abort, request
-from flask_login import current_user
 from flask_restx import Resource, marshal_with, reqparse
 
 import services
@@ -26,7 +25,7 @@ from controllers.console.wraps import (
 from extensions.ext_database import db
 from fields.member_fields import account_with_role_list_fields
 from libs.helper import extract_remote_ip
-from libs.login import login_required
+from libs.login import current_user, login_required
 from models.account import Account, TenantAccountRole
 from services.account_service import AccountService, RegisterService, TenantService
 from services.errors.account import AccountAlreadyInTenantError

api/controllers/console/workspace/plugin.py

@@ -540,18 +540,20 @@ class PluginFetchDynamicSelectOptionsApi(Resource):
         parser.add_argument("provider", type=str, required=True, location="args")
         parser.add_argument("action", type=str, required=True, location="args")
         parser.add_argument("parameter", type=str, required=True, location="args")
+        parser.add_argument("credential_id", type=str, required=False, location="args")
         parser.add_argument("provider_type", type=str, required=True, location="args")
         args = parser.parse_args()
 
         try:
             options = PluginParameterService.get_dynamic_select_options(
-                tenant_id,
-                user_id,
-                args["plugin_id"],
-                args["provider"],
-                args["action"],
-                args["parameter"],
-                args["provider_type"],
+                tenant_id=tenant_id,
+                user_id=user_id,
+                plugin_id=args["plugin_id"],
+                provider=args["provider"],
+                action=args["action"],
+                parameter=args["parameter"],
+                credential_id=args["credential_id"],
+                provider_type=args["provider_type"],
             )
         except PluginDaemonClientSideError as e:
             raise ValueError(e)

api/controllers/console/workspace/tool_providers.py

@@ -21,11 +21,13 @@ from core.mcp.auth.auth_provider import OAuthClientProvider
 from core.mcp.error import MCPAuthError, MCPError
 from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.impl.oauth import OAuthHandler
-from core.tools.entities.tool_entities import CredentialType
 from libs.helper import StrLen, alphanumeric, uuid_value
 from libs.login import login_required
 from models.provider_ids import ToolProviderID
+
+# from models.provider_ids import ToolProviderID
 from services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService

api/controllers/console/workspace/trigger_providers.py

@@ -0,0 +1,590 @@
+import logging
+
+from flask import make_response, redirect, request
+from flask_restx import Resource, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import BadRequest, Forbidden
+
+from configs import dify_config
+from controllers.console import api
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.model_runtime.utils.encoders import jsonable_encoder
+from core.plugin.entities.plugin_daemon import CredentialType
+from core.plugin.impl.oauth import OAuthHandler
+from core.trigger.entities.entities import SubscriptionBuilderUpdater
+from core.trigger.trigger_manager import TriggerManager
+from extensions.ext_database import db
+from libs.login import current_user, login_required
+from models.account import Account
+from models.provider_ids import TriggerProviderID
+from services.plugin.oauth_service import OAuthProxyService
+from services.trigger.trigger_provider_service import TriggerProviderService
+from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
+from services.trigger.workflow_plugin_trigger_service import WorkflowPluginTriggerService
+
+logger = logging.getLogger(__name__)
+
+
+class TriggerProviderIconApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+
+        return TriggerManager.get_trigger_plugin_icon(tenant_id=user.current_tenant_id, provider_id=provider)
+
+
+class TriggerProviderListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        """List all trigger providers for the current tenant"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        return jsonable_encoder(TriggerProviderService.list_trigger_providers(user.current_tenant_id))
+
+
+class TriggerProviderInfoApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """Get info for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        return jsonable_encoder(
+            TriggerProviderService.get_trigger_provider(user.current_tenant_id, TriggerProviderID(provider))
+        )
+
+
+class TriggerSubscriptionListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """List all trigger subscriptions for the current tenant's provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            return jsonable_encoder(
+                TriggerProviderService.list_trigger_provider_subscriptions(
+                    tenant_id=user.current_tenant_id, provider_id=TriggerProviderID(provider)
+                )
+            )
+        except Exception as e:
+            logger.exception("Error listing trigger providers", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderCreateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider):
+        """Add a new subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("credential_type", type=str, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            credential_type = CredentialType.of(args.get("credential_type") or CredentialType.UNAUTHORIZED.value)
+            subscription_builder = TriggerSubscriptionBuilderService.create_trigger_subscription_builder(
+                tenant_id=user.current_tenant_id,
+                user_id=user.id,
+                provider_id=TriggerProviderID(provider),
+                credential_type=credential_type,
+            )
+            return jsonable_encoder({"subscription_builder": subscription_builder})
+        except Exception as e:
+            logger.exception("Error adding provider credential", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderGetApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider, subscription_builder_id):
+        """Get a subscription instance for a trigger provider"""
+        return jsonable_encoder(
+            TriggerSubscriptionBuilderService.get_subscription_builder_by_id(subscription_builder_id)
+        )
+
+
+class TriggerSubscriptionBuilderVerifyApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider, subscription_builder_id):
+        """Verify a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        # The credentials of the subscription builder
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            # Use atomic update_and_verify to prevent race conditions
+            return TriggerSubscriptionBuilderService.update_and_verify_builder(
+                tenant_id=user.current_tenant_id,
+                user_id=user.id,
+                provider_id=TriggerProviderID(provider),
+                subscription_builder_id=subscription_builder_id,
+                subscription_builder_updater=SubscriptionBuilderUpdater(
+                    credentials=args.get("credentials", None),
+                ),
+            )
+        except Exception as e:
+            logger.exception("Error verifying provider credential", exc_info=e)
+            raise ValueError(str(e)) from e
+
+
+class TriggerSubscriptionBuilderUpdateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider, subscription_builder_id):
+        """Update a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+
+        parser = reqparse.RequestParser()
+        # The name of the subscription builder
+        parser.add_argument("name", type=str, required=False, nullable=True, location="json")
+        # The parameters of the subscription builder
+        parser.add_argument("parameters", type=dict, required=False, nullable=True, location="json")
+        # The properties of the subscription builder
+        parser.add_argument("properties", type=dict, required=False, nullable=True, location="json")
+        # The credentials of the subscription builder
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+        try:
+            return jsonable_encoder(
+                TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
+                    tenant_id=user.current_tenant_id,
+                    provider_id=TriggerProviderID(provider),
+                    subscription_builder_id=subscription_builder_id,
+                    subscription_builder_updater=SubscriptionBuilderUpdater(
+                        name=args.get("name", None),
+                        parameters=args.get("parameters", None),
+                        properties=args.get("properties", None),
+                        credentials=args.get("credentials", None),
+                    ),
+                )
+            )
+        except Exception as e:
+            logger.exception("Error updating provider credential", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderLogsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider, subscription_builder_id):
+        """Get the request logs for a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+
+        try:
+            logs = TriggerSubscriptionBuilderService.list_logs(subscription_builder_id)
+            return jsonable_encoder({"logs": [log.model_dump(mode="json") for log in logs]})
+        except Exception as e:
+            logger.exception("Error getting request logs for subscription builder", exc_info=e)
+            raise
+
+
+class TriggerSubscriptionBuilderBuildApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider, subscription_builder_id):
+        """Build a subscription instance for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        # The name of the subscription builder
+        parser.add_argument("name", type=str, required=False, nullable=True, location="json")
+        # The parameters of the subscription builder
+        parser.add_argument("parameters", type=dict, required=False, nullable=True, location="json")
+        # The properties of the subscription builder
+        parser.add_argument("properties", type=dict, required=False, nullable=True, location="json")
+        # The credentials of the subscription builder
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+        try:
+            # Use atomic update_and_build to prevent race conditions
+            TriggerSubscriptionBuilderService.update_and_build_builder(
+                tenant_id=user.current_tenant_id,
+                user_id=user.id,
+                provider_id=TriggerProviderID(provider),
+                subscription_builder_id=subscription_builder_id,
+                subscription_builder_updater=SubscriptionBuilderUpdater(
+                    name=args.get("name", None),
+                    parameters=args.get("parameters", None),
+                    properties=args.get("properties", None),
+                ),
+            )
+            return 200
+        except Exception as e:
+            logger.exception("Error building provider credential", exc_info=e)
+            raise ValueError(str(e)) from e
+
+
+class TriggerSubscriptionDeleteApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, subscription_id):
+        """Delete a subscription instance"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            with Session(db.engine) as session:
+                # Delete trigger provider subscription
+                TriggerProviderService.delete_trigger_provider(
+                    session=session,
+                    tenant_id=user.current_tenant_id,
+                    subscription_id=subscription_id,
+                )
+                # Delete plugin triggers
+                WorkflowPluginTriggerService.delete_plugin_trigger_by_subscription(
+                    session=session,
+                    tenant_id=user.current_tenant_id,
+                    subscription_id=subscription_id,
+                )
+                session.commit()
+            return {"result": "success"}
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error deleting provider credential", exc_info=e)
+            raise
+
+
+class TriggerOAuthAuthorizeApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """Initiate OAuth authorization flow for a trigger provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+
+        try:
+            provider_id = TriggerProviderID(provider)
+            plugin_id = provider_id.plugin_id
+            provider_name = provider_id.provider_name
+            tenant_id = user.current_tenant_id
+
+            # Get OAuth client configuration
+            oauth_client_params = TriggerProviderService.get_oauth_client(
+                tenant_id=tenant_id,
+                provider_id=provider_id,
+            )
+
+            if oauth_client_params is None:
+                raise Forbidden("No OAuth client configuration found for this trigger provider")
+
+            # Create subscription builder
+            subscription_builder = TriggerSubscriptionBuilderService.create_trigger_subscription_builder(
+                tenant_id=tenant_id,
+                user_id=user.id,
+                provider_id=provider_id,
+                credential_type=CredentialType.OAUTH2,
+            )
+
+            # Create OAuth handler and proxy context
+            oauth_handler = OAuthHandler()
+            context_id = OAuthProxyService.create_proxy_context(
+                user_id=user.id,
+                tenant_id=tenant_id,
+                plugin_id=plugin_id,
+                provider=provider_name,
+                extra_data={
+                    "subscription_builder_id": subscription_builder.id,
+                },
+            )
+
+            # Build redirect URI for callback
+            redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
+
+            # Get authorization URL
+            authorization_url_response = oauth_handler.get_authorization_url(
+                tenant_id=tenant_id,
+                user_id=user.id,
+                plugin_id=plugin_id,
+                provider=provider_name,
+                redirect_uri=redirect_uri,
+                system_credentials=oauth_client_params,
+            )
+
+            # Create response with cookie
+            response = make_response(
+                jsonable_encoder(
+                    {
+                        "authorization_url": authorization_url_response.authorization_url,
+                        "subscription_builder_id": subscription_builder.id,
+                        "subscription_builder": subscription_builder,
+                    }
+                )
+            )
+            response.set_cookie(
+                "context_id",
+                context_id,
+                httponly=True,
+                samesite="Lax",
+                max_age=OAuthProxyService.__MAX_AGE__,
+            )
+
+            return response
+
+        except Exception as e:
+            logger.exception("Error initiating OAuth flow", exc_info=e)
+            raise
+
+
+class TriggerOAuthCallbackApi(Resource):
+    @setup_required
+    def get(self, provider):
+        """Handle OAuth callback for trigger provider"""
+        context_id = request.cookies.get("context_id")
+        if not context_id:
+            raise Forbidden("context_id not found")
+
+        # Use and validate proxy context
+        context = OAuthProxyService.use_proxy_context(context_id)
+        if context is None:
+            raise Forbidden("Invalid context_id")
+
+        # Parse provider ID
+        provider_id = TriggerProviderID(provider)
+        plugin_id = provider_id.plugin_id
+        provider_name = provider_id.provider_name
+        user_id = context.get("user_id")
+        tenant_id = context.get("tenant_id")
+        subscription_builder_id = context.get("subscription_builder_id")
+
+        # Get OAuth client configuration
+        oauth_client_params = TriggerProviderService.get_oauth_client(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+        )
+
+        if oauth_client_params is None:
+            raise Forbidden("No OAuth client configuration found for this trigger provider")
+
+        # Get OAuth credentials from callback
+        oauth_handler = OAuthHandler()
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
+
+        credentials_response = oauth_handler.get_credentials(
+            tenant_id=tenant_id,
+            user_id=user_id,
+            plugin_id=plugin_id,
+            provider=provider_name,
+            redirect_uri=redirect_uri,
+            system_credentials=oauth_client_params,
+            request=request,
+        )
+
+        credentials = credentials_response.credentials
+        expires_at = credentials_response.expires_at
+
+        if not credentials:
+            raise Exception("Failed to get OAuth credentials")
+
+        # Update subscription builder
+        TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+            subscription_builder_id=subscription_builder_id,
+            subscription_builder_updater=SubscriptionBuilderUpdater(
+                credentials=credentials,
+                credential_expires_at=expires_at,
+            ),
+        )
+        # Redirect to OAuth callback page
+        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
+
+
+class TriggerOAuthClientManageApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        """Get OAuth client configuration for a provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            provider_id = TriggerProviderID(provider)
+
+            # Get custom OAuth client params if exists
+            custom_params = TriggerProviderService.get_custom_oauth_client_params(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+
+            # Check if custom client is enabled
+            is_custom_enabled = TriggerProviderService.is_oauth_custom_client_enabled(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+
+            # Check if there's a system OAuth client
+            system_client = TriggerProviderService.get_oauth_client(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+            provider_controller = TriggerManager.get_trigger_provider(user.current_tenant_id, provider_id)
+            redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/trigger/callback"
+            return jsonable_encoder(
+                {
+                    "configured": bool(custom_params or system_client),
+                    "oauth_client_schema": provider_controller.get_oauth_client_schema(),
+                    "custom_configured": bool(custom_params),
+                    "custom_enabled": is_custom_enabled,
+                    "redirect_uri": redirect_uri,
+                    "params": custom_params or {},
+                }
+            )
+
+        except Exception as e:
+            logger.exception("Error getting OAuth client", exc_info=e)
+            raise
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider):
+        """Configure custom OAuth client for a provider"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("client_params", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("enabled", type=bool, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            provider_id = TriggerProviderID(provider)
+            return TriggerProviderService.save_custom_oauth_client_params(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+                client_params=args.get("client_params"),
+                enabled=args.get("enabled"),
+            )
+
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error configuring OAuth client", exc_info=e)
+            raise
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, provider):
+        """Remove custom OAuth client configuration"""
+        user = current_user
+        assert isinstance(user, Account)
+        assert user.current_tenant_id is not None
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        try:
+            provider_id = TriggerProviderID(provider)
+
+            return TriggerProviderService.delete_custom_oauth_client_params(
+                tenant_id=user.current_tenant_id,
+                provider_id=provider_id,
+            )
+        except ValueError as e:
+            raise BadRequest(str(e))
+        except Exception as e:
+            logger.exception("Error removing OAuth client", exc_info=e)
+            raise
+
+
+# Trigger Subscription
+api.add_resource(TriggerProviderIconApi, "/workspaces/current/trigger-provider/<path:provider>/icon")
+api.add_resource(TriggerProviderListApi, "/workspaces/current/triggers")
+api.add_resource(TriggerProviderInfoApi, "/workspaces/current/trigger-provider/<path:provider>/info")
+api.add_resource(TriggerSubscriptionListApi, "/workspaces/current/trigger-provider/<path:provider>/subscriptions/list")
+api.add_resource(
+    TriggerSubscriptionDeleteApi,
+    "/workspaces/current/trigger-provider/<path:subscription_id>/subscriptions/delete",
+)
+
+# Trigger Subscription Builder
+api.add_resource(
+    TriggerSubscriptionBuilderCreateApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/create",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderGetApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderUpdateApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/update/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderVerifyApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/verify/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderBuildApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/build/<path:subscription_builder_id>",
+)
+api.add_resource(
+    TriggerSubscriptionBuilderLogsApi,
+    "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/logs/<path:subscription_builder_id>",
+)
+
+
+# OAuth
+api.add_resource(
+    TriggerOAuthAuthorizeApi, "/workspaces/current/trigger-provider/<path:provider>/subscriptions/oauth/authorize"
+)
+api.add_resource(TriggerOAuthCallbackApi, "/oauth/plugin/<path:provider>/trigger/callback")
+api.add_resource(TriggerOAuthClientManageApi, "/workspaces/current/trigger-provider/<path:provider>/oauth/client")

api/controllers/console/workspace/workspace.py

@@ -1,7 +1,6 @@
 import logging
 
 from flask import request
-from flask_login import current_user
 from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse
 from sqlalchemy import select
 from werkzeug.exceptions import Unauthorized
@@ -24,7 +23,7 @@ from controllers.console.wraps import (
 )
 from extensions.ext_database import db
 from libs.helper import TimestampField
-from libs.login import login_required
+from libs.login import current_user, login_required
 from models.account import Account, Tenant, TenantStatus
 from services.account_service import TenantService
 from services.feature_service import FeatureService

api/controllers/console/wraps.py

@@ -7,13 +7,13 @@ from functools import wraps
 from typing import ParamSpec, TypeVar
 
 from flask import abort, request
-from flask_login import current_user
 
 from configs import dify_config
 from controllers.console.workspace.error import AccountNotInitializedError
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
-from models.account import AccountStatus
+from libs.login import current_user
+from models.account import Account, AccountStatus
 from models.dataset import RateLimitLog
 from models.model import DifySetup
 from services.feature_service import FeatureService, LicenseStatus
@@ -25,11 +25,16 @@ P = ParamSpec("P")
 R = TypeVar("R")
 
 
+def _current_account() -> Account:
+    assert isinstance(current_user, Account)
+    return current_user
+
+
 def account_initialization_required(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         # check account initialization
-        account = current_user
+        account = _current_account()
 
         if account.status == AccountStatus.UNINITIALIZED:
             raise AccountNotInitializedError()
@@ -75,7 +80,9 @@ def only_edition_self_hosted(view: Callable[P, R]):
 def cloud_edition_billing_enabled(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
-        features = FeatureService.get_features(current_user.current_tenant_id)
+        account = _current_account()
+        assert account.current_tenant_id is not None
+        features = FeatureService.get_features(account.current_tenant_id)
         if not features.billing.enabled:
             abort(403, "Billing feature is not enabled.")
         return view(*args, **kwargs)
@@ -87,7 +94,10 @@ def cloud_edition_billing_resource_check(resource: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
-            features = FeatureService.get_features(current_user.current_tenant_id)
+            account = _current_account()
+            assert account.current_tenant_id is not None
+            tenant_id = account.current_tenant_id
+            features = FeatureService.get_features(tenant_id)
             if features.billing.enabled:
                 members = features.members
                 apps = features.apps
@@ -128,7 +138,9 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
     def interceptor(view: Callable[P, R]):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
-            features = FeatureService.get_features(current_user.current_tenant_id)
+            account = _current_account()
+            assert account.current_tenant_id is not None
+            features = FeatureService.get_features(account.current_tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
                     if features.billing.subscription.plan == "sandbox":
@@ -151,10 +163,13 @@ def cloud_edition_billing_rate_limit_check(resource: str):
         @wraps(view)
         def decorated(*args: P.args, **kwargs: P.kwargs):
             if resource == "knowledge":
-                knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(current_user.current_tenant_id)
+                account = _current_account()
+                assert account.current_tenant_id is not None
+                tenant_id = account.current_tenant_id
+                knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(tenant_id)
                 if knowledge_rate_limit.enabled:
                     current_time = int(time.time() * 1000)
-                    key = f"rate_limit_{current_user.current_tenant_id}"
+                    key = f"rate_limit_{tenant_id}"
 
                     redis_client.zadd(key, {current_time: current_time})
 
@@ -165,7 +180,7 @@ def cloud_edition_billing_rate_limit_check(resource: str):
                     if request_count > knowledge_rate_limit.limit:
                         # add ratelimit record
                         rate_limit_log = RateLimitLog(
-                            tenant_id=current_user.current_tenant_id,
+                            tenant_id=tenant_id,
                             subscription_plan=knowledge_rate_limit.subscription_plan,
                             operation="knowledge",
                         )
@@ -185,14 +200,17 @@ def cloud_utm_record(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
         with contextlib.suppress(Exception):
-            features = FeatureService.get_features(current_user.current_tenant_id)
+            account = _current_account()
+            assert account.current_tenant_id is not None
+            tenant_id = account.current_tenant_id
+            features = FeatureService.get_features(tenant_id)
 
             if features.billing.enabled:
                 utm_info = request.cookies.get("utm_info")
 
                 if utm_info:
                     utm_info_dict: dict = json.loads(utm_info)
-                    OperationService.record_utm(current_user.current_tenant_id, utm_info_dict)
+                    OperationService.record_utm(tenant_id, utm_info_dict)
 
         return view(*args, **kwargs)
 
@@ -271,7 +289,9 @@ def enable_change_email(view: Callable[P, R]):
 def is_allow_transfer_owner(view: Callable[P, R]):
     @wraps(view)
     def decorated(*args: P.args, **kwargs: P.kwargs):
-        features = FeatureService.get_features(current_user.current_tenant_id)
+        account = _current_account()
+        assert account.current_tenant_id is not None
+        features = FeatureService.get_features(account.current_tenant_id)
         if features.is_allow_transfer_workspace:
             return view(*args, **kwargs)
 
@@ -284,7 +304,9 @@ def is_allow_transfer_owner(view: Callable[P, R]):
 def knowledge_pipeline_publish_enabled(view):
     @wraps(view)
     def decorated(*args, **kwargs):
-        features = FeatureService.get_features(current_user.current_tenant_id)
+        account = _current_account()
+        assert account.current_tenant_id is not None
+        features = FeatureService.get_features(account.current_tenant_id)
         if features.knowledge_pipeline.publish_enabled:
             return view(*args, **kwargs)
         abort(403)

api/controllers/inner_api/plugin/wraps.py

@@ -25,8 +25,8 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     As a result, it could only be considered as an end user id.
     """
     if not user_id:
-        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
-    is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
+        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
+    is_anonymous = user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID
     try:
         with Session(db.engine) as session:
             user_model = None
@@ -85,7 +85,7 @@ def get_user_tenant(view: Callable[P, R] | None = None):
                 raise ValueError("tenant_id is required")
 
             if not user_id:
-                user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
+                user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
 
             try:
                 tenant_model = (

api/controllers/mcp/mcp.py

@@ -9,10 +9,9 @@ from controllers.console.app.mcp_server import AppMCPServerStatus
 from controllers.mcp import mcp_ns
 from core.app.app_config.entities import VariableEntity
 from core.mcp import types as mcp_types
-from core.mcp.server.streamable_http import handle_mcp_request
 from extensions.ext_database import db
 from libs import helper
-from models.model import App, AppMCPServer, AppMode, EndUser
+from models.model import App, AppMCPServer, AppMode
 
 
 class MCPRequestError(Exception):
@@ -195,50 +194,6 @@ class MCPAppApi(Resource):
             except ValidationError as e:
                 raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
 
-    def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str, session: Session) -> EndUser | None:
-        """Get end user from existing session - optimized query"""
-        return (
-            session.query(EndUser)
-            .where(EndUser.tenant_id == tenant_id)
-            .where(EndUser.session_id == mcp_server_id)
-            .where(EndUser.type == "mcp")
-            .first()
-        )
-
-    def _create_end_user(
-        self, client_name: str, tenant_id: str, app_id: str, mcp_server_id: str, session: Session
-    ) -> EndUser:
-        """Create end user in existing session"""
-        end_user = EndUser(
-            tenant_id=tenant_id,
-            app_id=app_id,
-            type="mcp",
-            name=client_name,
-            session_id=mcp_server_id,
-        )
-        session.add(end_user)
-        session.flush()  # Use flush instead of commit to keep transaction open
-        session.refresh(end_user)
-        return end_user
-
-    def _handle_mcp_request(
-        self,
-        app: App,
-        mcp_server: AppMCPServer,
-        mcp_request: mcp_types.ClientRequest,
-        user_input_form: list[VariableEntity],
-        session: Session,
-        request_id: Union[int, str],
-    ) -> mcp_types.JSONRPCResponse | mcp_types.JSONRPCError | None:
-        """Handle MCP request and return response"""
-        end_user = self._retrieve_end_user(mcp_server.tenant_id, mcp_server.id, session)
-
-        if not end_user and isinstance(mcp_request.root, mcp_types.InitializeRequest):
-            client_info = mcp_request.root.params.clientInfo
-            client_name = f"{client_info.name}@{client_info.version}"
-            # Commit the session before creating end user to avoid transaction conflicts
-            session.commit()
-            with Session(db.engine, expire_on_commit=False) as create_session, create_session.begin():
-                end_user = self._create_end_user(client_name, app.tenant_id, app.id, mcp_server.id, create_session)
-
-        return handle_mcp_request(app, mcp_request, user_input_form, mcp_server, end_user, request_id)
+        mcp_server_handler = MCPServerStreamableHTTPRequestHandler(app, request, converted_user_input_form)
+        response = mcp_server_handler.handle()
+        return helper.compact_generate_response(response)

api/controllers/service_api/wraps.py

@@ -313,7 +313,7 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: str | None =
     Create or update session terminal based on user ID.
     """
     if not user_id:
-        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID.value
+        user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
 
     with Session(db.engine, expire_on_commit=False) as session:
         end_user = (
@@ -332,7 +332,7 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: str | None =
                 tenant_id=app_model.tenant_id,
                 app_id=app_model.id,
                 type="service_api",
-                is_anonymous=user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID.value,
+                is_anonymous=user_id == DefaultEndUserSessionID.DEFAULT_SESSION_ID,
                 session_id=user_id,
             )
             session.add(end_user)

api/controllers/trigger/__init__.py

@@ -0,0 +1,12 @@
+from flask import Blueprint
+
+# Create trigger blueprint
+bp = Blueprint("trigger", __name__, url_prefix="/triggers")
+
+# Import routes after blueprint creation to avoid circular imports
+from . import trigger, webhook
+
+__all__ = [
+    "trigger",
+    "webhook",
+]

api/controllers/trigger/trigger.py

@@ -0,0 +1,43 @@
+import logging
+import re
+
+from flask import jsonify, request
+from werkzeug.exceptions import NotFound
+
+from controllers.trigger import bp
+from services.trigger.trigger_service import TriggerService
+from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService
+
+logger = logging.getLogger(__name__)
+
+UUID_PATTERN = r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
+UUID_MATCHER = re.compile(UUID_PATTERN)
+
+
+@bp.route("/plugin/<string:endpoint_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
+def trigger_endpoint(endpoint_id: str):
+    """
+    Handle endpoint trigger calls.
+    """
+    # endpoint_id must be UUID
+    if not UUID_MATCHER.match(endpoint_id):
+        raise NotFound("Invalid endpoint ID")
+    handling_chain = [
+        TriggerService.process_endpoint,
+        TriggerSubscriptionBuilderService.process_builder_validation_endpoint,
+    ]
+    response = None
+    try:
+        for handler in handling_chain:
+            response = handler(endpoint_id, request)
+            if response:
+                break
+        if not response:
+            logger.error("Endpoint not found for {endpoint_id}")
+            return jsonify({"error": "Endpoint not found"}), 404
+        return response
+    except ValueError as e:
+        raise NotFound(str(e))
+    except Exception as e:
+        logger.exception("Webhook processing failed for {endpoint_id}")
+        return jsonify({"error": "Internal server error", "message": str(e)}), 500

api/controllers/trigger/webhook.py

@@ -0,0 +1,104 @@
+import logging
+import time
+
+from flask import jsonify
+from werkzeug.exceptions import NotFound, RequestEntityTooLarge
+
+from controllers.trigger import bp
+from services.trigger.trigger_debug_service import TriggerDebugService, WebhookDebugEvent
+from services.trigger.webhook_service import WebhookService
+
+logger = logging.getLogger(__name__)
+
+
+def _prepare_webhook_execution(webhook_id: str, is_debug: bool = False):
+    """Fetch trigger context, extract request data, and validate payload using unified processing.
+
+    Args:
+        webhook_id: The webhook ID to process
+        is_debug: If True, skip status validation for debug mode
+    """
+    webhook_trigger, workflow, node_config = WebhookService.get_webhook_trigger_and_workflow(
+        webhook_id, is_debug=is_debug
+    )
+
+    try:
+        # Use new unified extraction and validation
+        webhook_data = WebhookService.extract_and_validate_webhook_data(webhook_trigger, node_config)
+        return webhook_trigger, workflow, node_config, webhook_data, None
+    except ValueError as e:
+        # Fall back to raw extraction for error reporting
+        webhook_data = WebhookService.extract_webhook_data(webhook_trigger)
+        return webhook_trigger, workflow, node_config, webhook_data, str(e)
+
+
+@bp.route("/webhook/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
+def handle_webhook(webhook_id: str):
+    """
+    Handle webhook trigger calls.
+
+    This endpoint receives webhook calls and processes them according to the
+    configured webhook trigger settings.
+    """
+    try:
+        webhook_trigger, workflow, node_config, webhook_data, error = _prepare_webhook_execution(webhook_id)
+        if error:
+            return jsonify({"error": "Bad Request", "message": error}), 400
+
+        # Process webhook call (send to Celery)
+        WebhookService.trigger_workflow_execution(webhook_trigger, webhook_data, workflow)
+
+        # Return configured response
+        response_data, status_code = WebhookService.generate_webhook_response(node_config)
+        return jsonify(response_data), status_code
+
+    except ValueError as e:
+        raise NotFound(str(e))
+    except RequestEntityTooLarge:
+        raise
+    except Exception as e:
+        logger.exception("Webhook processing failed for %s", webhook_id)
+        return jsonify({"error": "Internal server error", "message": str(e)}), 500
+
+
+@bp.route("/webhook-debug/<string:webhook_id>", methods=["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"])
+def handle_webhook_debug(webhook_id: str):
+    """Handle webhook debug calls without triggering production workflow execution."""
+    try:
+        webhook_trigger, _, node_config, webhook_data, error = _prepare_webhook_execution(webhook_id, is_debug=True)
+        if error:
+            return jsonify({"error": "Bad Request", "message": error}), 400
+
+        workflow_inputs = WebhookService.build_workflow_inputs(webhook_data)
+
+        # Generate pool key and dispatch debug event
+        pool_key: str = WebhookDebugEvent.build_pool_key(
+            tenant_id=webhook_trigger.tenant_id,
+            app_id=webhook_trigger.app_id,
+            node_id=webhook_trigger.node_id,
+        )
+        event = WebhookDebugEvent(
+            request_id=f"webhook_debug_{webhook_trigger.webhook_id}_{int(time.time() * 1000)}",
+            timestamp=int(time.time()),
+            node_id=webhook_trigger.node_id,
+            payload={
+                "inputs": workflow_inputs,
+                "webhook_data": webhook_data,
+                "method": webhook_data.get("method"),
+            },
+        )
+        TriggerDebugService.dispatch(
+            tenant_id=webhook_trigger.tenant_id,
+            event=event,
+            pool_key=pool_key,
+        )
+        response_data, status_code = WebhookService.generate_webhook_response(node_config)
+        return jsonify(response_data), status_code
+
+    except ValueError as e:
+        raise NotFound(str(e))
+    except RequestEntityTooLarge:
+        raise
+    except Exception as e:
+        logger.exception("Webhook debug processing failed for %s", webhook_id)
+        return jsonify({"error": "Internal server error", "message": str(e)}), 500

api/core/app/app_config/easy_ui_based_app/dataset/manager.py

@@ -197,12 +197,12 @@ class DatasetConfigManager:
 
         # strategy
         if "strategy" not in config["agent_mode"] or not config["agent_mode"].get("strategy"):
-            config["agent_mode"]["strategy"] = PlanningStrategy.ROUTER.value
+            config["agent_mode"]["strategy"] = PlanningStrategy.ROUTER
 
         has_datasets = False
         if config.get("agent_mode", {}).get("strategy") in {
-            PlanningStrategy.ROUTER.value,
-            PlanningStrategy.REACT_ROUTER.value,
+            PlanningStrategy.ROUTER,
+            PlanningStrategy.REACT_ROUTER,
         }:
             for tool in config.get("agent_mode", {}).get("tools", []):
                 key = list(tool.keys())[0]

api/core/app/app_config/easy_ui_based_app/model_config/converter.py

@@ -68,9 +68,13 @@ class ModelConfigConverter:
         # get model mode
         model_mode = model_config.mode
         if not model_mode:
-            model_mode = LLMMode.CHAT.value
+            model_mode = LLMMode.CHAT
             if model_schema and model_schema.model_properties.get(ModelPropertyKey.MODE):
-                model_mode = LLMMode(model_schema.model_properties[ModelPropertyKey.MODE]).value
+                try:
+                    model_mode = LLMMode(model_schema.model_properties[ModelPropertyKey.MODE])
+                except ValueError:
+                    # Fall back to CHAT mode if the stored value is invalid
+                    model_mode = LLMMode.CHAT
 
         if not model_schema:
             raise ValueError(f"Model {model_name} not exist.")

api/core/app/app_config/easy_ui_based_app/prompt_template/manager.py

@@ -100,7 +100,7 @@ class PromptTemplateConfigManager:
             if config["model"]["mode"] not in model_mode_vals:
                 raise ValueError(f"model.mode must be in {model_mode_vals} when prompt_type is advanced")
 
-            if app_mode == AppMode.CHAT and config["model"]["mode"] == ModelMode.COMPLETION.value:
+            if app_mode == AppMode.CHAT and config["model"]["mode"] == ModelMode.COMPLETION:
                 user_prefix = config["completion_prompt_config"]["conversation_histories_role"]["user_prefix"]
                 assistant_prefix = config["completion_prompt_config"]["conversation_histories_role"]["assistant_prefix"]
 
@@ -110,7 +110,7 @@ class PromptTemplateConfigManager:
                 if not assistant_prefix:
                     config["completion_prompt_config"]["conversation_histories_role"]["assistant_prefix"] = "Assistant"
 
-            if config["model"]["mode"] == ModelMode.CHAT.value:
+            if config["model"]["mode"] == ModelMode.CHAT:
                 prompt_list = config["chat_prompt_config"]["prompt"]
 
                 if len(prompt_list) > 10:

api/core/app/apps/agent_chat/app_config_manager.py

@@ -186,7 +186,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
             raise ValueError("enabled in agent_mode must be of boolean type")
 
         if not agent_mode.get("strategy"):
-            agent_mode["strategy"] = PlanningStrategy.ROUTER.value
+            agent_mode["strategy"] = PlanningStrategy.ROUTER
 
         if agent_mode["strategy"] not in [member.value for member in list(PlanningStrategy.__members__.values())]:
             raise ValueError("strategy in agent_mode must be in the specified strategy list")

api/core/app/apps/agent_chat/app_runner.py

@@ -198,9 +198,9 @@ class AgentChatAppRunner(AppRunner):
         # start agent runner
         if agent_entity.strategy == AgentEntity.Strategy.CHAIN_OF_THOUGHT:
             # check LLM mode
-            if model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.CHAT.value:
+            if model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.CHAT:
                 runner_cls = CotChatAgentRunner
-            elif model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.COMPLETION.value:
+            elif model_schema.model_properties.get(ModelPropertyKey.MODE) == LLMMode.COMPLETION:
                 runner_cls = CotCompletionAgentRunner
             else:
                 raise ValueError(f"Invalid LLM mode: {model_schema.model_properties.get(ModelPropertyKey.MODE)}")

api/core/app/apps/common/workflow_response_converter.py

@@ -38,6 +38,7 @@ from core.file import FILE_MODEL_IDENTITY, File
 from core.plugin.impl.datasource import PluginDatasourceManager
 from core.tools.entities.tool_entities import ToolProviderType
 from core.tools.tool_manager import ToolManager
+from core.trigger.trigger_manager import TriggerManager
 from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities import WorkflowExecution, WorkflowNodeExecution
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
@@ -181,6 +182,11 @@ class WorkflowResponseConverter:
             response.data.extras["icon"] = provider_entity.declaration.identity.generate_datasource_icon_url(
                 self._application_generate_entity.app_config.tenant_id
             )
+        elif event.node_type == NodeType.TRIGGER_PLUGIN:
+            response.data.extras["icon"] = TriggerManager.get_trigger_plugin_icon(
+                self._application_generate_entity.app_config.tenant_id,
+                event.provider_id,
+            )
 
         return response
 

api/core/app/apps/pipeline/pipeline_runner.py

@@ -229,8 +229,8 @@ class PipelineRunner(WorkflowBasedAppRunner):
             workflow_id=workflow.id,
             graph_config=graph_config,
             user_id=self.application_generate_entity.user_id,
-            user_from=UserFrom.ACCOUNT.value,
-            invoke_from=InvokeFrom.SERVICE_API.value,
+            user_from=UserFrom.ACCOUNT,
+            invoke_from=InvokeFrom.SERVICE_API,
             call_depth=0,
         )
 

api/core/app/apps/workflow/app_generator.py

@@ -3,7 +3,7 @@ import logging
 import threading
 import uuid
 from collections.abc import Generator, Mapping, Sequence
-from typing import Any, Literal, Union, overload
+from typing import Any, Literal, Optional, Union, overload
 
 from flask import Flask, current_app
 from pydantic import ValidationError
@@ -53,6 +53,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[True],
         call_depth: int,
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Generator[Mapping | str, None, None]: ...
 
     @overload
@@ -66,6 +68,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: Literal[False],
         call_depth: int,
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Mapping[str, Any]: ...
 
     @overload
@@ -79,6 +83,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool,
         call_depth: int,
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]: ...
 
     def generate(
@@ -91,6 +97,8 @@ class WorkflowAppGenerator(BaseAppGenerator):
         invoke_from: InvokeFrom,
         streaming: bool = True,
         call_depth: int = 0,
+        triggered_from: Optional[WorkflowRunTriggeredFrom] = None,
+        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[Mapping | str, None, None]]:
         files: Sequence[Mapping[str, Any]] = args.get("files") or []
 
@@ -126,17 +134,20 @@ class WorkflowAppGenerator(BaseAppGenerator):
             **extract_external_trace_id_from_args(args),
         }
         workflow_run_id = str(uuid.uuid4())
+        if triggered_from in (WorkflowRunTriggeredFrom.DEBUGGING, WorkflowRunTriggeredFrom.APP_RUN):
+            # start node get inputs
+            inputs = self._prepare_user_inputs(
+                user_inputs=inputs,
+                variables=app_config.variables,
+                tenant_id=app_model.tenant_id,
+                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
+            )
         # init application generate entity
         application_generate_entity = WorkflowAppGenerateEntity(
             task_id=str(uuid.uuid4()),
             app_config=app_config,
             file_upload_config=file_extra_config,
-            inputs=self._prepare_user_inputs(
-                user_inputs=inputs,
-                variables=app_config.variables,
-                tenant_id=app_model.tenant_id,
-                strict_type_validation=True if invoke_from == InvokeFrom.SERVICE_API else False,
-            ),
+            inputs=inputs,
             files=list(system_files),
             user_id=user.id,
             stream=streaming,
@@ -155,7 +166,10 @@ class WorkflowAppGenerator(BaseAppGenerator):
         # Create session factory
         session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
         # Create workflow execution(aka workflow run) repository
-        if invoke_from == InvokeFrom.DEBUGGER:
+        if triggered_from is not None:
+            # Use explicitly provided triggered_from (for async triggers)
+            workflow_triggered_from = triggered_from
+        elif invoke_from == InvokeFrom.DEBUGGER:
             workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
         else:
             workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
@@ -182,6 +196,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
             workflow_execution_repository=workflow_execution_repository,
             workflow_node_execution_repository=workflow_node_execution_repository,
             streaming=streaming,
+            root_node_id=root_node_id,
         )
 
     def _generate(
@@ -196,6 +211,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         streaming: bool = True,
         variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
+        root_node_id: Optional[str] = None,
     ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
         """
         Generate App response.
@@ -231,6 +247,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 "queue_manager": queue_manager,
                 "context": context,
                 "variable_loader": variable_loader,
+                "root_node_id": root_node_id,
             },
         )
 
@@ -424,6 +441,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
         queue_manager: AppQueueManager,
         context: contextvars.Context,
         variable_loader: VariableLoader,
+        root_node_id: Optional[str] = None,
     ) -> None:
         """
         Generate worker in a new thread.
@@ -465,6 +483,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
                 variable_loader=variable_loader,
                 workflow=workflow,
                 system_user_id=system_user_id,
+                root_node_id=root_node_id,
             )
 
             try:

api/core/app/apps/workflow/app_runner.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import cast
+from typing import Optional, cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfig
@@ -34,6 +34,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         variable_loader: VariableLoader,
         workflow: Workflow,
         system_user_id: str,
+        root_node_id: Optional[str] = None,
     ):
         super().__init__(
             queue_manager=queue_manager,
@@ -43,6 +44,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
         self.application_generate_entity = application_generate_entity
         self._workflow = workflow
         self._sys_user_id = system_user_id
+        self._root_node_id = root_node_id
 
     def run(self):
         """
@@ -88,6 +90,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 workflow_id=self._workflow.id,
                 tenant_id=self._workflow.tenant_id,
                 user_id=self.application_generate_entity.user_id,
+                root_node_id=self._root_node_id,
             )
 
         # RUN WORKFLOW

api/core/app/apps/workflow_app_runner.py

@@ -1,6 +1,6 @@
 import time
 from collections.abc import Mapping
-from typing import Any, cast
+from typing import Any, Optional, cast
 
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -80,6 +80,7 @@ class WorkflowBasedAppRunner:
         workflow_id: str = "",
         tenant_id: str = "",
         user_id: str = "",
+        root_node_id: Optional[str] = None,
     ) -> Graph:
         """
         Init graph
@@ -100,8 +101,8 @@ class WorkflowBasedAppRunner:
             workflow_id=workflow_id,
             graph_config=graph_config,
             user_id=user_id,
-            user_from=UserFrom.ACCOUNT.value,
-            invoke_from=InvokeFrom.SERVICE_API.value,
+            user_from=UserFrom.ACCOUNT,
+            invoke_from=InvokeFrom.SERVICE_API,
             call_depth=0,
         )
 
@@ -113,7 +114,7 @@ class WorkflowBasedAppRunner:
         )
 
         # init graph
-        graph = Graph.init(graph_config=graph_config, node_factory=node_factory)
+        graph = Graph.init(graph_config=graph_config, node_factory=node_factory, root_node_id=root_node_id)
 
         if not graph:
             raise ValueError("graph not found in workflow")
@@ -244,8 +245,8 @@ class WorkflowBasedAppRunner:
             workflow_id=workflow.id,
             graph_config=graph_config,
             user_id="",
-            user_from=UserFrom.ACCOUNT.value,
-            invoke_from=InvokeFrom.SERVICE_API.value,
+            user_from=UserFrom.ACCOUNT,
+            invoke_from=InvokeFrom.SERVICE_API,
             call_depth=0,
         )
 

api/core/datasource/entities/api_entities.py

@@ -49,7 +49,7 @@ class DatasourceProviderApiEntity(BaseModel):
         for datasource in datasources:
             if datasource.get("parameters"):
                 for parameter in datasource.get("parameters"):
-                    if parameter.get("type") == DatasourceParameter.DatasourceParameterType.SYSTEM_FILES.value:
+                    if parameter.get("type") == DatasourceParameter.DatasourceParameterType.SYSTEM_FILES:
                         parameter["type"] = "files"
         # -------------
 

api/core/datasource/entities/datasource_entities.py

@@ -1,5 +1,5 @@
 import enum
-from enum import Enum
+from enum import StrEnum
 from typing import Any
 
 from pydantic import BaseModel, Field, ValidationInfo, field_validator
@@ -54,16 +54,16 @@ class DatasourceParameter(PluginParameter):
         removes TOOLS_SELECTOR from PluginParameterType
         """
 
-        STRING = PluginParameterType.STRING.value
-        NUMBER = PluginParameterType.NUMBER.value
-        BOOLEAN = PluginParameterType.BOOLEAN.value
-        SELECT = PluginParameterType.SELECT.value
-        SECRET_INPUT = PluginParameterType.SECRET_INPUT.value
-        FILE = PluginParameterType.FILE.value
-        FILES = PluginParameterType.FILES.value
+        STRING = PluginParameterType.STRING
+        NUMBER = PluginParameterType.NUMBER
+        BOOLEAN = PluginParameterType.BOOLEAN
+        SELECT = PluginParameterType.SELECT
+        SECRET_INPUT = PluginParameterType.SECRET_INPUT
+        FILE = PluginParameterType.FILE
+        FILES = PluginParameterType.FILES
 
         # deprecated, should not use.
-        SYSTEM_FILES = PluginParameterType.SYSTEM_FILES.value
+        SYSTEM_FILES = PluginParameterType.SYSTEM_FILES
 
         def as_normal_type(self):
             return as_normal_type(self)
@@ -218,7 +218,7 @@ class DatasourceLabel(BaseModel):
     icon: str = Field(..., description="The icon of the tool")
 
 
-class DatasourceInvokeFrom(Enum):
+class DatasourceInvokeFrom(StrEnum):
     """
     Enum class for datasource invoke
     """

api/core/entities/parameter_entities.py

@@ -14,6 +14,7 @@ class CommonParameterType(StrEnum):
     APP_SELECTOR = "app-selector"
     MODEL_SELECTOR = "model-selector"
     TOOLS_SELECTOR = "array[tools]"
+    CHECKBOX = "checkbox"
     ANY = auto()
 
     # Dynamic select parameter

api/core/entities/provider_configuration.py

@@ -207,7 +207,7 @@ class ProviderConfiguration(BaseModel):
         """
         stmt = select(Provider).where(
             Provider.tenant_id == self.tenant_id,
-            Provider.provider_type == ProviderType.CUSTOM.value,
+            Provider.provider_type == ProviderType.CUSTOM,
             Provider.provider_name.in_(self._get_provider_names()),
         )
 
@@ -458,7 +458,7 @@ class ProviderConfiguration(BaseModel):
                     provider_record = Provider(
                         tenant_id=self.tenant_id,
                         provider_name=self.provider.provider,
-                        provider_type=ProviderType.CUSTOM.value,
+                        provider_type=ProviderType.CUSTOM,
                         is_valid=True,
                         credential_id=new_record.id,
                     )
@@ -1414,7 +1414,7 @@ class ProviderConfiguration(BaseModel):
         """
         secret_input_form_variables = []
         for credential_form_schema in credential_form_schemas:
-            if credential_form_schema.type.value == FormType.SECRET_INPUT.value:
+            if credential_form_schema.type == FormType.SECRET_INPUT:
                 secret_input_form_variables.append(credential_form_schema.variable)
 
         return secret_input_form_variables

api/core/entities/provider_entities.py

@@ -1,5 +1,5 @@
 from enum import StrEnum, auto
-from typing import Union
+from typing import Optional, Union
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -84,9 +84,9 @@ class SystemConfiguration(BaseModel):
     """
 
     enabled: bool
-    current_quota_type: ProviderQuotaType | None = None
+    current_quota_type: Optional[ProviderQuotaType] = None
     quota_configurations: list[QuotaConfiguration] = []
-    credentials: dict | None = None
+    credentials: Optional[dict] = None
 
 
 class CustomProviderConfiguration(BaseModel):
@@ -95,8 +95,8 @@ class CustomProviderConfiguration(BaseModel):
     """
 
     credentials: dict
-    current_credential_id: str | None = None
-    current_credential_name: str | None = None
+    current_credential_id: Optional[str] = None
+    current_credential_name: Optional[str] = None
     available_credentials: list[CredentialConfiguration] = []
 
 
@@ -107,9 +107,9 @@ class CustomModelConfiguration(BaseModel):
 
     model: str
     model_type: ModelType
-    credentials: dict | None = None
-    current_credential_id: str | None = None
-    current_credential_name: str | None = None
+    credentials: dict | None
+    current_credential_id: Optional[str] = None
+    current_credential_name: Optional[str] = None
     available_model_credentials: list[CredentialConfiguration] = []
     unadded_to_model_list: bool | None = False
 
@@ -131,7 +131,7 @@ class CustomConfiguration(BaseModel):
     Model class for provider custom configuration.
     """
 
-    provider: CustomProviderConfiguration | None = None
+    provider: Optional[CustomProviderConfiguration] = None
     models: list[CustomModelConfiguration] = []
     can_added_models: list[UnaddedModelConfiguration] = []
 
@@ -207,6 +207,7 @@ class ProviderConfig(BasicProviderConfig):
     required: bool = False
     default: Union[int, str, float, bool] | None = None
     options: list[Option] | None = None
+    multiple: bool | None = False
     label: I18nObject | None = None
     help: I18nObject | None = None
     url: str | None = None

api/core/extension/api_based_extension_requestor.py

@@ -1,13 +1,13 @@
 from typing import cast
 
-import requests
+import httpx
 
 from configs import dify_config
 from models.api_based_extension import APIBasedExtensionPoint
 
 
 class APIBasedExtensionRequestor:
-    timeout: tuple[int, int] = (5, 60)
+    timeout: httpx.Timeout = httpx.Timeout(60.0, connect=5.0)
     """timeout for request connect and read"""
 
     def __init__(self, api_endpoint: str, api_key: str):
@@ -27,25 +27,23 @@ class APIBasedExtensionRequestor:
         url = self.api_endpoint
 
         try:
-            # proxy support for security
-            proxies = None
+            mounts: dict[str, httpx.BaseTransport] | None = None
             if dify_config.SSRF_PROXY_HTTP_URL and dify_config.SSRF_PROXY_HTTPS_URL:
-                proxies = {
-                    "http": dify_config.SSRF_PROXY_HTTP_URL,
-                    "https": dify_config.SSRF_PROXY_HTTPS_URL,
+                mounts = {
+                    "http://": httpx.HTTPTransport(proxy=dify_config.SSRF_PROXY_HTTP_URL),
+                    "https://": httpx.HTTPTransport(proxy=dify_config.SSRF_PROXY_HTTPS_URL),
                 }
 
-            response = requests.request(
-                method="POST",
-                url=url,
-                json={"point": point.value, "params": params},
-                headers=headers,
-                timeout=self.timeout,
-                proxies=proxies,
-            )
-        except requests.Timeout:
+            with httpx.Client(mounts=mounts, timeout=self.timeout) as client:
+                response = client.request(
+                    method="POST",
+                    url=url,
+                    json={"point": point.value, "params": params},
+                    headers=headers,
+                )
+        except httpx.TimeoutException:
             raise ValueError("request timeout")
-        except requests.ConnectionError:
+        except httpx.RequestError:
             raise ValueError("request connection error")
 
         if response.status_code != 200:

api/core/helper/name_generator.py

@@ -3,7 +3,7 @@ import re
 from collections.abc import Sequence
 from typing import Any
 
-from core.tools.entities.tool_entities import CredentialType
+from core.plugin.entities.plugin_daemon import CredentialType
 
 logger = logging.getLogger(__name__)
 

api/core/helper/provider_encryption.py

@@ -0,0 +1,128 @@
+import contextlib
+from copy import deepcopy
+from typing import Any, Optional, Protocol
+
+from core.entities.provider_entities import BasicProviderConfig
+from core.helper import encrypter
+
+
+class ProviderConfigCache(Protocol):
+    """
+    Interface for provider configuration cache operations
+    """
+
+    def get(self) -> Optional[dict]:
+        """Get cached provider configuration"""
+        ...
+
+    def set(self, config: dict[str, Any]) -> None:
+        """Cache provider configuration"""
+        ...
+
+    def delete(self) -> None:
+        """Delete cached provider configuration"""
+        ...
+
+
+class ProviderConfigEncrypter:
+    tenant_id: str
+    config: list[BasicProviderConfig]
+    provider_config_cache: ProviderConfigCache
+
+    def __init__(
+        self,
+        tenant_id: str,
+        config: list[BasicProviderConfig],
+        provider_config_cache: ProviderConfigCache,
+    ):
+        self.tenant_id = tenant_id
+        self.config = config
+        self.provider_config_cache = provider_config_cache
+
+    def _deep_copy(self, data: dict[str, str]) -> dict[str, str]:
+        """
+        deep copy data
+        """
+        return deepcopy(data)
+
+    def encrypt(self, data: dict[str, str]) -> dict[str, str]:
+        """
+        encrypt tool credentials with tenant id
+
+        return a deep copy of credentials with encrypted values
+        """
+        data = self._deep_copy(data)
+
+        # get fields need to be decrypted
+        fields = dict[str, BasicProviderConfig]()
+        for credential in self.config:
+            fields[credential.name] = credential
+
+        for field_name, field in fields.items():
+            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
+                if field_name in data:
+                    encrypted = encrypter.encrypt_token(self.tenant_id, data[field_name] or "")
+                    data[field_name] = encrypted
+
+        return data
+
+    def mask_credentials(self, data: dict[str, Any]) -> dict[str, Any]:
+        """
+        mask credentials
+
+        return a deep copy of credentials with masked values
+        """
+        data = self._deep_copy(data)
+
+        # get fields need to be decrypted
+        fields = dict[str, BasicProviderConfig]()
+        for credential in self.config:
+            fields[credential.name] = credential
+
+        for field_name, field in fields.items():
+            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
+                if field_name in data:
+                    if len(data[field_name]) > 6:
+                        data[field_name] = (
+                            data[field_name][:2] + "*" * (len(data[field_name]) - 4) + data[field_name][-2:]
+                        )
+                    else:
+                        data[field_name] = "*" * len(data[field_name])
+
+        return data
+
+    def mask_tool_credentials(self, data: dict[str, Any]) -> dict[str, Any]:
+        return self.mask_credentials(data)
+
+    def decrypt(self, data: dict[str, str]) -> dict[str, Any]:
+        """
+        decrypt tool credentials with tenant id
+
+        return a deep copy of credentials with decrypted values
+        """
+        cached_credentials = self.provider_config_cache.get()
+        if cached_credentials:
+            return cached_credentials
+
+        data = self._deep_copy(data)
+        # get fields need to be decrypted
+        fields = dict[str, BasicProviderConfig]()
+        for credential in self.config:
+            fields[credential.name] = credential
+
+        for field_name, field in fields.items():
+            if field.type == BasicProviderConfig.Type.SECRET_INPUT:
+                if field_name in data:
+                    with contextlib.suppress(Exception):
+                        # if the value is None or empty string, skip decrypt
+                        if not data[field_name]:
+                            continue
+
+                        data[field_name] = encrypter.decrypt_token(self.tenant_id, data[field_name])
+
+        self.provider_config_cache.set(data)
+        return data
+
+
+def create_provider_encrypter(tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache):
+    return ProviderConfigEncrypter(tenant_id=tenant_id, config=config, provider_config_cache=cache), cache

api/core/indexing_runner.py

@@ -343,7 +343,7 @@ class IndexingRunner:
 
             if file_detail:
                 extract_setting = ExtractSetting(
-                    datasource_type=DatasourceType.FILE.value,
+                    datasource_type=DatasourceType.FILE,
                     upload_file=file_detail,
                     document_model=dataset_document.doc_form,
                 )
@@ -356,7 +356,7 @@ class IndexingRunner:
             ):
                 raise ValueError("no notion import info found")
             extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.NOTION.value,
+                datasource_type=DatasourceType.NOTION,
                 notion_info=NotionInfo.model_validate(
                     {
                         "credential_id": data_source_info["credential_id"],
@@ -379,7 +379,7 @@ class IndexingRunner:
             ):
                 raise ValueError("no website import info found")
             extract_setting = ExtractSetting(
-                datasource_type=DatasourceType.WEBSITE.value,
+                datasource_type=DatasourceType.WEBSITE,
                 website_info=WebsiteInfo.model_validate(
                     {
                         "provider": data_source_info["provider"],

api/core/llm_generator/output_parser/structured_output.py

@@ -224,8 +224,8 @@ def _handle_native_json_schema(
 
     # Set appropriate response format if required by the model
     for rule in rules:
-        if rule.name == "response_format" and ResponseFormat.JSON_SCHEMA.value in rule.options:
-            model_parameters["response_format"] = ResponseFormat.JSON_SCHEMA.value
+        if rule.name == "response_format" and ResponseFormat.JSON_SCHEMA in rule.options:
+            model_parameters["response_format"] = ResponseFormat.JSON_SCHEMA
 
     return model_parameters
 
@@ -239,10 +239,10 @@ def _set_response_format(model_parameters: dict, rules: list):
     """
     for rule in rules:
         if rule.name == "response_format":
-            if ResponseFormat.JSON.value in rule.options:
-                model_parameters["response_format"] = ResponseFormat.JSON.value
-            elif ResponseFormat.JSON_OBJECT.value in rule.options:
-                model_parameters["response_format"] = ResponseFormat.JSON_OBJECT.value
+            if ResponseFormat.JSON in rule.options:
+                model_parameters["response_format"] = ResponseFormat.JSON
+            elif ResponseFormat.JSON_OBJECT in rule.options:
+                model_parameters["response_format"] = ResponseFormat.JSON_OBJECT
 
 
 def _handle_prompt_based_schema(

api/core/model_runtime/entities/provider_entities.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from enum import Enum, StrEnum, auto
+from enum import StrEnum, auto
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
@@ -7,7 +7,7 @@ from core.model_runtime.entities.common_entities import I18nObject
 from core.model_runtime.entities.model_entities import AIModelEntity, ModelType
 
 
-class ConfigurateMethod(Enum):
+class ConfigurateMethod(StrEnum):
     """
     Enum class for configurate method of provider model.
     """

api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -213,9 +213,9 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                     node_metadata.update(json.loads(node_execution.execution_metadata))
 
                 # Determine the correct span kind based on node type
-                span_kind = OpenInferenceSpanKindValues.CHAIN.value
+                span_kind = OpenInferenceSpanKindValues.CHAIN
                 if node_execution.node_type == "llm":
-                    span_kind = OpenInferenceSpanKindValues.LLM.value
+                    span_kind = OpenInferenceSpanKindValues.LLM
                     provider = process_data.get("model_provider")
                     model = process_data.get("model_name")
                     if provider:
@@ -230,18 +230,18 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                         node_metadata["prompt_tokens"] = usage_data.get("prompt_tokens", 0)
                         node_metadata["completion_tokens"] = usage_data.get("completion_tokens", 0)
                 elif node_execution.node_type == "dataset_retrieval":
-                    span_kind = OpenInferenceSpanKindValues.RETRIEVER.value
+                    span_kind = OpenInferenceSpanKindValues.RETRIEVER
                 elif node_execution.node_type == "tool":
-                    span_kind = OpenInferenceSpanKindValues.TOOL.value
+                    span_kind = OpenInferenceSpanKindValues.TOOL
                 else:
-                    span_kind = OpenInferenceSpanKindValues.CHAIN.value
+                    span_kind = OpenInferenceSpanKindValues.CHAIN
 
                 node_span = self.tracer.start_span(
                     name=node_execution.node_type,
                     attributes={
                         SpanAttributes.INPUT_VALUE: node_execution.inputs or "{}",
                         SpanAttributes.OUTPUT_VALUE: node_execution.outputs or "{}",
-                        SpanAttributes.OPENINFERENCE_SPAN_KIND: span_kind,
+                        SpanAttributes.OPENINFERENCE_SPAN_KIND: span_kind.value,
                         SpanAttributes.METADATA: json.dumps(node_metadata, ensure_ascii=False),
                         SpanAttributes.SESSION_ID: trace_info.conversation_id or "",
                     },

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -73,7 +73,7 @@ class LangFuseDataTrace(BaseTraceInstance):
 
         if trace_info.message_id:
             trace_id = trace_info.trace_id or trace_info.message_id
-            name = TraceTaskName.MESSAGE_TRACE.value
+            name = TraceTaskName.MESSAGE_TRACE
             trace_data = LangfuseTrace(
                 id=trace_id,
                 user_id=user_id,
@@ -88,7 +88,7 @@ class LangFuseDataTrace(BaseTraceInstance):
             self.add_trace(langfuse_trace_data=trace_data)
             workflow_span_data = LangfuseSpan(
                 id=trace_info.workflow_run_id,
-                name=TraceTaskName.WORKFLOW_TRACE.value,
+                name=TraceTaskName.WORKFLOW_TRACE,
                 input=dict(trace_info.workflow_run_inputs),
                 output=dict(trace_info.workflow_run_outputs),
                 trace_id=trace_id,
@@ -103,7 +103,7 @@ class LangFuseDataTrace(BaseTraceInstance):
             trace_data = LangfuseTrace(
                 id=trace_id,
                 user_id=user_id,
-                name=TraceTaskName.WORKFLOW_TRACE.value,
+                name=TraceTaskName.WORKFLOW_TRACE,
                 input=dict(trace_info.workflow_run_inputs),
                 output=dict(trace_info.workflow_run_outputs),
                 metadata=metadata,
@@ -253,7 +253,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         trace_data = LangfuseTrace(
             id=trace_id,
             user_id=user_id,
-            name=TraceTaskName.MESSAGE_TRACE.value,
+            name=TraceTaskName.MESSAGE_TRACE,
             input={
                 "message": trace_info.inputs,
                 "files": file_list,
@@ -303,7 +303,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         span_data = LangfuseSpan(
-            name=TraceTaskName.MODERATION_TRACE.value,
+            name=TraceTaskName.MODERATION_TRACE,
             input=trace_info.inputs,
             output={
                 "action": trace_info.action,
@@ -331,7 +331,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         )
 
         generation_data = LangfuseGeneration(
-            name=TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
+            name=TraceTaskName.SUGGESTED_QUESTION_TRACE,
             input=trace_info.inputs,
             output=str(trace_info.suggested_question),
             trace_id=trace_info.trace_id or trace_info.message_id,
@@ -349,7 +349,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         dataset_retrieval_span_data = LangfuseSpan(
-            name=TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
+            name=TraceTaskName.DATASET_RETRIEVAL_TRACE,
             input=trace_info.inputs,
             output={"documents": trace_info.documents},
             trace_id=trace_info.trace_id or trace_info.message_id,
@@ -377,7 +377,7 @@ class LangFuseDataTrace(BaseTraceInstance):
 
     def generate_name_trace(self, trace_info: GenerateNameTraceInfo):
         name_generation_trace_data = LangfuseTrace(
-            name=TraceTaskName.GENERATE_NAME_TRACE.value,
+            name=TraceTaskName.GENERATE_NAME_TRACE,
             input=trace_info.inputs,
             output=trace_info.outputs,
             user_id=trace_info.tenant_id,
@@ -388,7 +388,7 @@ class LangFuseDataTrace(BaseTraceInstance):
         self.add_trace(langfuse_trace_data=name_generation_trace_data)
 
         name_generation_span_data = LangfuseSpan(
-            name=TraceTaskName.GENERATE_NAME_TRACE.value,
+            name=TraceTaskName.GENERATE_NAME_TRACE,
             input=trace_info.inputs,
             output=trace_info.outputs,
             trace_id=trace_info.conversation_id,

api/core/ops/langsmith_trace/langsmith_trace.py

@@ -81,7 +81,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if trace_info.message_id:
             message_run = LangSmithRunModel(
                 id=trace_info.message_id,
-                name=TraceTaskName.MESSAGE_TRACE.value,
+                name=TraceTaskName.MESSAGE_TRACE,
                 inputs=dict(trace_info.workflow_run_inputs),
                 outputs=dict(trace_info.workflow_run_outputs),
                 run_type=LangSmithRunType.chain,
@@ -110,7 +110,7 @@ class LangSmithDataTrace(BaseTraceInstance):
             file_list=trace_info.file_list,
             total_tokens=trace_info.total_tokens,
             id=trace_info.workflow_run_id,
-            name=TraceTaskName.WORKFLOW_TRACE.value,
+            name=TraceTaskName.WORKFLOW_TRACE,
             inputs=dict(trace_info.workflow_run_inputs),
             run_type=LangSmithRunType.tool,
             start_time=trace_info.workflow_data.created_at,
@@ -271,7 +271,7 @@ class LangSmithDataTrace(BaseTraceInstance):
             output_tokens=trace_info.answer_tokens,
             total_tokens=trace_info.total_tokens,
             id=message_id,
-            name=TraceTaskName.MESSAGE_TRACE.value,
+            name=TraceTaskName.MESSAGE_TRACE,
             inputs=trace_info.inputs,
             run_type=LangSmithRunType.chain,
             start_time=trace_info.start_time,
@@ -327,7 +327,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         langsmith_run = LangSmithRunModel(
-            name=TraceTaskName.MODERATION_TRACE.value,
+            name=TraceTaskName.MODERATION_TRACE,
             inputs=trace_info.inputs,
             outputs={
                 "action": trace_info.action,
@@ -362,7 +362,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if message_data is None:
             return
         suggested_question_run = LangSmithRunModel(
-            name=TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
+            name=TraceTaskName.SUGGESTED_QUESTION_TRACE,
             inputs=trace_info.inputs,
             outputs=trace_info.suggested_question,
             run_type=LangSmithRunType.tool,
@@ -391,7 +391,7 @@ class LangSmithDataTrace(BaseTraceInstance):
         if trace_info.message_data is None:
             return
         dataset_retrieval_run = LangSmithRunModel(
-            name=TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
+            name=TraceTaskName.DATASET_RETRIEVAL_TRACE,
             inputs=trace_info.inputs,
             outputs={"documents": trace_info.documents},
             run_type=LangSmithRunType.retriever,
@@ -447,7 +447,7 @@ class LangSmithDataTrace(BaseTraceInstance):
 
     def generate_name_trace(self, trace_info: GenerateNameTraceInfo):
         name_run = LangSmithRunModel(
-            name=TraceTaskName.GENERATE_NAME_TRACE.value,
+            name=TraceTaskName.GENERATE_NAME_TRACE,
             inputs=trace_info.inputs,
             outputs=trace_info.outputs,
             run_type=LangSmithRunType.tool,

api/core/ops/opik_trace/opik_trace.py

@@ -108,7 +108,7 @@ class OpikDataTrace(BaseTraceInstance):
 
             trace_data = {
                 "id": opik_trace_id,
-                "name": TraceTaskName.MESSAGE_TRACE.value,
+                "name": TraceTaskName.MESSAGE_TRACE,
                 "start_time": trace_info.start_time,
                 "end_time": trace_info.end_time,
                 "metadata": workflow_metadata,
@@ -125,7 +125,7 @@ class OpikDataTrace(BaseTraceInstance):
                 "id": root_span_id,
                 "parent_span_id": None,
                 "trace_id": opik_trace_id,
-                "name": TraceTaskName.WORKFLOW_TRACE.value,
+                "name": TraceTaskName.WORKFLOW_TRACE,
                 "input": wrap_dict("input", trace_info.workflow_run_inputs),
                 "output": wrap_dict("output", trace_info.workflow_run_outputs),
                 "start_time": trace_info.start_time,
@@ -138,7 +138,7 @@ class OpikDataTrace(BaseTraceInstance):
         else:
             trace_data = {
                 "id": opik_trace_id,
-                "name": TraceTaskName.MESSAGE_TRACE.value,
+                "name": TraceTaskName.MESSAGE_TRACE,
                 "start_time": trace_info.start_time,
                 "end_time": trace_info.end_time,
                 "metadata": workflow_metadata,
@@ -290,7 +290,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         trace_data = {
             "id": prepare_opik_uuid(trace_info.start_time, dify_trace_id),
-            "name": TraceTaskName.MESSAGE_TRACE.value,
+            "name": TraceTaskName.MESSAGE_TRACE,
             "start_time": trace_info.start_time,
             "end_time": trace_info.end_time,
             "metadata": wrap_metadata(metadata),
@@ -329,7 +329,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.MODERATION_TRACE.value,
+            "name": TraceTaskName.MODERATION_TRACE,
             "type": "tool",
             "start_time": start_time,
             "end_time": trace_info.end_time or trace_info.message_data.updated_at,
@@ -355,7 +355,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.SUGGESTED_QUESTION_TRACE.value,
+            "name": TraceTaskName.SUGGESTED_QUESTION_TRACE,
             "type": "tool",
             "start_time": start_time,
             "end_time": trace_info.end_time or message_data.updated_at,
@@ -375,7 +375,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": prepare_opik_uuid(start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.DATASET_RETRIEVAL_TRACE.value,
+            "name": TraceTaskName.DATASET_RETRIEVAL_TRACE,
             "type": "tool",
             "start_time": start_time,
             "end_time": trace_info.end_time or trace_info.message_data.updated_at,
@@ -405,7 +405,7 @@ class OpikDataTrace(BaseTraceInstance):
     def generate_name_trace(self, trace_info: GenerateNameTraceInfo):
         trace_data = {
             "id": prepare_opik_uuid(trace_info.start_time, trace_info.trace_id or trace_info.message_id),
-            "name": TraceTaskName.GENERATE_NAME_TRACE.value,
+            "name": TraceTaskName.GENERATE_NAME_TRACE,
             "start_time": trace_info.start_time,
             "end_time": trace_info.end_time,
             "metadata": wrap_metadata(trace_info.metadata),
@@ -420,7 +420,7 @@ class OpikDataTrace(BaseTraceInstance):
 
         span_data = {
             "trace_id": trace.id,
-            "name": TraceTaskName.GENERATE_NAME_TRACE.value,
+            "name": TraceTaskName.GENERATE_NAME_TRACE,
             "start_time": trace_info.start_time,
             "end_time": trace_info.end_time,
             "metadata": wrap_metadata(trace_info.metadata),

api/core/ops/weave_trace/weave_trace.py

@@ -104,7 +104,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
             message_run = WeaveTraceModel(
                 id=trace_info.message_id,
-                op=str(TraceTaskName.MESSAGE_TRACE.value),
+                op=str(TraceTaskName.MESSAGE_TRACE),
                 inputs=dict(trace_info.workflow_run_inputs),
                 outputs=dict(trace_info.workflow_run_outputs),
                 total_tokens=trace_info.total_tokens,
@@ -126,7 +126,7 @@ class WeaveDataTrace(BaseTraceInstance):
             file_list=trace_info.file_list,
             total_tokens=trace_info.total_tokens,
             id=trace_info.workflow_run_id,
-            op=str(TraceTaskName.WORKFLOW_TRACE.value),
+            op=str(TraceTaskName.WORKFLOW_TRACE),
             inputs=dict(trace_info.workflow_run_inputs),
             outputs=dict(trace_info.workflow_run_outputs),
             attributes=workflow_attributes,
@@ -253,7 +253,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         message_run = WeaveTraceModel(
             id=trace_id,
-            op=str(TraceTaskName.MESSAGE_TRACE.value),
+            op=str(TraceTaskName.MESSAGE_TRACE),
             input_tokens=trace_info.message_tokens,
             output_tokens=trace_info.answer_tokens,
             total_tokens=trace_info.total_tokens,
@@ -300,7 +300,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         moderation_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.MODERATION_TRACE.value),
+            op=str(TraceTaskName.MODERATION_TRACE),
             inputs=trace_info.inputs,
             outputs={
                 "action": trace_info.action,
@@ -330,7 +330,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         suggested_question_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.SUGGESTED_QUESTION_TRACE.value),
+            op=str(TraceTaskName.SUGGESTED_QUESTION_TRACE),
             inputs=trace_info.inputs,
             outputs=trace_info.suggested_question,
             attributes=attributes,
@@ -355,7 +355,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         dataset_retrieval_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.DATASET_RETRIEVAL_TRACE.value),
+            op=str(TraceTaskName.DATASET_RETRIEVAL_TRACE),
             inputs=trace_info.inputs,
             outputs={"documents": trace_info.documents},
             attributes=attributes,
@@ -397,7 +397,7 @@ class WeaveDataTrace(BaseTraceInstance):
 
         name_run = WeaveTraceModel(
             id=str(uuid.uuid4()),
-            op=str(TraceTaskName.GENERATE_NAME_TRACE.value),
+            op=str(TraceTaskName.GENERATE_NAME_TRACE),
             inputs=trace_info.inputs,
             outputs=trace_info.outputs,
             attributes=attributes,

api/core/plugin/backwards_invocation/node.py

@@ -52,7 +52,7 @@ class PluginNodeBackwardsInvocation(BaseBackwardsInvocation):
             instruction=instruction,  # instruct with variables are not supported
         )
         node_data_dict = node_data.model_dump()
-        node_data_dict["type"] = NodeType.PARAMETER_EXTRACTOR.value
+        node_data_dict["type"] = NodeType.PARAMETER_EXTRACTOR
         execution = workflow_service.run_free_workflow_node(
             node_data_dict,
             tenant_id=tenant_id,

api/core/plugin/entities/parameters.py

@@ -39,7 +39,7 @@ class PluginParameterType(StrEnum):
     TOOLS_SELECTOR = CommonParameterType.TOOLS_SELECTOR
     ANY = CommonParameterType.ANY
     DYNAMIC_SELECT = CommonParameterType.DYNAMIC_SELECT
-
+    CHECKBOX = CommonParameterType.CHECKBOX
     # deprecated, should not use.
     SYSTEM_FILES = CommonParameterType.SYSTEM_FILES
 
@@ -94,6 +94,7 @@ def as_normal_type(typ: StrEnum):
     if typ.value in {
         PluginParameterType.SECRET_INPUT,
         PluginParameterType.SELECT,
+        PluginParameterType.CHECKBOX,
     }:
         return "string"
     return typ.value
@@ -102,7 +103,13 @@ def as_normal_type(typ: StrEnum):
 def cast_parameter_value(typ: StrEnum, value: Any, /):
     try:
         match typ.value:
-            case PluginParameterType.STRING | PluginParameterType.SECRET_INPUT | PluginParameterType.SELECT:
+            case (
+                PluginParameterType.STRING
+                | PluginParameterType.SECRET_INPUT
+                | PluginParameterType.SELECT
+                | PluginParameterType.CHECKBOX
+                | PluginParameterType.DYNAMIC_SELECT
+            ):
                 if value is None:
                     return ""
                 else:

api/core/plugin/entities/plugin.py

@@ -1,7 +1,7 @@
 import datetime
 from collections.abc import Mapping
 from enum import StrEnum, auto
-from typing import Any
+from typing import Any, Optional
 
 from packaging.version import InvalidVersion, Version
 from pydantic import BaseModel, Field, field_validator, model_validator
@@ -13,6 +13,7 @@ from core.plugin.entities.base import BasePluginEntity
 from core.plugin.entities.endpoint import EndpointProviderDeclaration
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntity
+from core.trigger.entities.entities import TriggerProviderEntity
 
 
 class PluginInstallationSource(StrEnum):
@@ -63,6 +64,7 @@ class PluginCategory(StrEnum):
     Extension = auto()
     AgentStrategy = "agent-strategy"
     Datasource = "datasource"
+    Trigger = "trigger"
 
 
 class PluginDeclaration(BaseModel):
@@ -71,6 +73,7 @@ class PluginDeclaration(BaseModel):
         models: list[str] | None = Field(default_factory=list[str])
         endpoints: list[str] | None = Field(default_factory=list[str])
         datasources: list[str] | None = Field(default_factory=list[str])
+        triggers: Optional[list[str]] = Field(default_factory=list[str])
 
     class Meta(BaseModel):
         minimum_dify_version: str | None = Field(default=None)
@@ -106,6 +109,7 @@ class PluginDeclaration(BaseModel):
     endpoint: EndpointProviderDeclaration | None = None
     agent_strategy: AgentStrategyProviderEntity | None = None
     datasource: DatasourceProviderEntity | None = None
+    trigger: Optional[TriggerProviderEntity] = None
     meta: Meta
 
     @field_validator("version")
@@ -129,6 +133,8 @@ class PluginDeclaration(BaseModel):
             values["category"] = PluginCategory.Datasource
         elif values.get("agent_strategy"):
             values["category"] = PluginCategory.AgentStrategy
+        elif values.get("trigger"):
+            values["category"] = PluginCategory.Trigger
         else:
             values["category"] = PluginCategory.Extension
         return values

api/core/plugin/entities/plugin_daemon.py

@@ -1,3 +1,4 @@
+import enum
 from collections.abc import Mapping, Sequence
 from datetime import datetime
 from enum import StrEnum
@@ -14,6 +15,7 @@ from core.plugin.entities.parameters import PluginParameterOption
 from core.plugin.entities.plugin import PluginDeclaration, PluginEntity
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntityWithPlugin
+from core.trigger.entities.entities import TriggerProviderEntity
 
 T = TypeVar("T", bound=(BaseModel | dict | list | bool | str))
 
@@ -205,3 +207,48 @@ class PluginListResponse(BaseModel):
 
 class PluginDynamicSelectOptionsResponse(BaseModel):
     options: Sequence[PluginParameterOption] = Field(description="The options of the dynamic select.")
+
+
+class PluginTriggerProviderEntity(BaseModel):
+    provider: str
+    plugin_unique_identifier: str
+    plugin_id: str
+    declaration: TriggerProviderEntity
+
+
+class CredentialType(enum.StrEnum):
+    API_KEY = "api-key"
+    OAUTH2 = "oauth2"
+    UNAUTHORIZED = "unauthorized"
+
+    def get_name(self):
+        if self == CredentialType.API_KEY:
+            return "API KEY"
+        elif self == CredentialType.OAUTH2:
+            return "AUTH"
+        elif self == CredentialType.UNAUTHORIZED:
+            return "UNAUTHORIZED"
+        else:
+            return self.value.replace("-", " ").upper()
+
+    def is_editable(self):
+        return self == CredentialType.API_KEY
+
+    def is_validate_allowed(self):
+        return self == CredentialType.API_KEY
+
+    @classmethod
+    def values(cls):
+        return [item.value for item in cls]
+
+    @classmethod
+    def of(cls, credential_type: str) -> "CredentialType":
+        type_name = credential_type.lower()
+        if type_name in {"api-key", "api_key"}:
+            return cls.API_KEY
+        elif type_name in {"oauth2", "oauth"}:
+            return cls.OAUTH2
+        elif type_name == "unauthorized":
+            return cls.UNAUTHORIZED
+        else:
+            raise ValueError(f"Invalid credential type: {credential_type}")

api/core/plugin/entities/request.py

@@ -1,5 +1,7 @@
-from typing import Any, Literal
+from collections.abc import Mapping
+from typing import Any, Literal, Optional
 
+from flask import Response
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from core.entities.provider_entities import BasicProviderConfig
@@ -83,13 +85,13 @@ class RequestInvokeLLM(BaseRequestInvokeModel):
             raise ValueError("prompt_messages must be a list")
 
         for i in range(len(v)):
-            if v[i]["role"] == PromptMessageRole.USER.value:
+            if v[i]["role"] == PromptMessageRole.USER:
                 v[i] = UserPromptMessage.model_validate(v[i])
-            elif v[i]["role"] == PromptMessageRole.ASSISTANT.value:
+            elif v[i]["role"] == PromptMessageRole.ASSISTANT:
                 v[i] = AssistantPromptMessage.model_validate(v[i])
-            elif v[i]["role"] == PromptMessageRole.SYSTEM.value:
+            elif v[i]["role"] == PromptMessageRole.SYSTEM:
                 v[i] = SystemPromptMessage.model_validate(v[i])
-            elif v[i]["role"] == PromptMessageRole.TOOL.value:
+            elif v[i]["role"] == PromptMessageRole.TOOL:
                 v[i] = ToolPromptMessage.model_validate(v[i])
             else:
                 v[i] = PromptMessage.model_validate(v[i])
@@ -237,3 +239,30 @@ class RequestFetchAppInfo(BaseModel):
     """
 
     app_id: str
+
+
+class TriggerInvokeEventResponse(BaseModel):
+    variables: Mapping[str, Any] = Field(default_factory=dict)
+    cancelled: Optional[bool] = False
+
+
+class PluginTriggerDispatchResponse(BaseModel):
+    events: list[str]
+    raw_http_response: str
+
+
+class TriggerSubscriptionResponse(BaseModel):
+    subscription: dict[str, Any]
+
+
+class TriggerValidateProviderCredentialsResponse(BaseModel):
+    result: bool
+
+
+class TriggerDispatchResponse:
+    events: list[str]
+    response: Response
+
+    def __init__(self, events: list[str], response: Response):
+        self.events = events
+        self.response = response

api/core/plugin/impl/base.py

@@ -2,11 +2,10 @@ import inspect
 import json
 import logging
 from collections.abc import Callable, Generator
-from typing import TypeVar
+from typing import Any, TypeVar
 
-import requests
+import httpx
 from pydantic import BaseModel
-from requests.exceptions import HTTPError
 from yarl import URL
 
 from configs import dify_config
@@ -47,29 +46,56 @@ class BasePluginClient:
         data: bytes | dict | str | None = None,
         params: dict | None = None,
         files: dict | None = None,
-        stream: bool = False,
-    ) -> requests.Response:
+    ) -> httpx.Response:
         """
         Make a request to the plugin daemon inner API.
         """
-        url = plugin_daemon_inner_api_baseurl / path
-        headers = headers or {}
-        headers["X-Api-Key"] = dify_config.PLUGIN_DAEMON_KEY
-        headers["Accept-Encoding"] = "gzip, deflate, br"
+        url, headers, prepared_data, params, files = self._prepare_request(path, headers, data, params, files)
 
-        if headers.get("Content-Type") == "application/json" and isinstance(data, dict):
-            data = json.dumps(data)
+        request_kwargs: dict[str, Any] = {
+            "method": method,
+            "url": url,
+            "headers": headers,
+            "params": params,
+            "files": files,
+        }
+        if isinstance(prepared_data, dict):
+            request_kwargs["data"] = prepared_data
+        elif prepared_data is not None:
+            request_kwargs["content"] = prepared_data
 
         try:
-            response = requests.request(
-                method=method, url=str(url), headers=headers, data=data, params=params, stream=stream, files=files
-            )
-        except requests.ConnectionError:
+            response = httpx.request(**request_kwargs)
+        except httpx.RequestError:
             logger.exception("Request to Plugin Daemon Service failed")
             raise PluginDaemonInnerError(code=-500, message="Request to Plugin Daemon Service failed")
 
         return response
 
+    def _prepare_request(
+        self,
+        path: str,
+        headers: dict | None,
+        data: bytes | dict | str | None,
+        params: dict | None,
+        files: dict | None,
+    ) -> tuple[str, dict, bytes | dict | str | None, dict | None, dict | None]:
+        url = plugin_daemon_inner_api_baseurl / path
+        prepared_headers = dict(headers or {})
+        prepared_headers["X-Api-Key"] = dify_config.PLUGIN_DAEMON_KEY
+        prepared_headers.setdefault("Accept-Encoding", "gzip, deflate, br")
+
+        prepared_data: bytes | dict | str | None = (
+            data if isinstance(data, (bytes, str, dict)) or data is None else None
+        )
+        if isinstance(data, dict):
+            if prepared_headers.get("Content-Type") == "application/json":
+                prepared_data = json.dumps(data)
+            else:
+                prepared_data = data
+
+        return str(url), prepared_headers, prepared_data, params, files
+
     def _stream_request(
         self,
         method: str,
@@ -78,17 +104,38 @@ class BasePluginClient:
         headers: dict | None = None,
         data: bytes | dict | None = None,
         files: dict | None = None,
-    ) -> Generator[bytes, None, None]:
+    ) -> Generator[str, None, None]:
         """
         Make a stream request to the plugin daemon inner API
         """
-        response = self._request(method, path, headers, data, params, files, stream=True)
-        for line in response.iter_lines(chunk_size=1024 * 8):
-            line = line.decode("utf-8").strip()
-            if line.startswith("data:"):
-                line = line[5:].strip()
-            if line:
-                yield line
+        url, headers, prepared_data, params, files = self._prepare_request(path, headers, data, params, files)
+
+        stream_kwargs: dict[str, Any] = {
+            "method": method,
+            "url": url,
+            "headers": headers,
+            "params": params,
+            "files": files,
+        }
+        if isinstance(prepared_data, dict):
+            stream_kwargs["data"] = prepared_data
+        elif prepared_data is not None:
+            stream_kwargs["content"] = prepared_data
+
+        try:
+            with httpx.stream(**stream_kwargs) as response:
+                for raw_line in response.iter_lines():
+                    if raw_line is None:
+                        continue
+                    line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line
+                    line = line.strip()
+                    if line.startswith("data:"):
+                        line = line[5:].strip()
+                    if line:
+                        yield line
+        except httpx.RequestError:
+            logger.exception("Stream request to Plugin Daemon Service failed")
+            raise PluginDaemonInnerError(code=-500, message="Request to Plugin Daemon Service failed")
 
     def _stream_request_with_model(
         self,
@@ -139,7 +186,7 @@ class BasePluginClient:
         try:
             response = self._request(method, path, headers, data, params, files)
             response.raise_for_status()
-        except HTTPError as e:
+        except httpx.HTTPStatusError as e:
             logger.exception("Failed to request plugin daemon, status: %s, url: %s", e.response.status_code, path)
             raise e
         except Exception as e:
@@ -208,7 +255,7 @@ class BasePluginClient:
                     except Exception:
                         raise PluginDaemonInnerError(code=rep.code, message=rep.message)
 
-                    logger.error("Error in stream reponse for plugin %s", rep.__dict__)
+                    logger.error("Error in stream response for plugin %s", rep.__dict__)
                     self._handle_plugin_daemon_error(error.error_type, error.message)
                 raise ValueError(f"plugin daemon: {rep.message}, code: {rep.code}")
             if rep.data is None:

api/core/plugin/impl/dynamic_select.py

@@ -15,6 +15,7 @@ class DynamicSelectClient(BasePluginClient):
         provider: str,
         action: str,
         credentials: Mapping[str, Any],
+        credential_type: str,
         parameter: str,
     ) -> PluginDynamicSelectOptionsResponse:
         """
@@ -29,6 +30,7 @@ class DynamicSelectClient(BasePluginClient):
                 "data": {
                     "provider": GenericProviderID(provider).provider_name,
                     "credentials": credentials,
+                    "credential_type": credential_type,
                     "provider_action": action,
                     "parameter": parameter,
                 },

api/core/plugin/impl/tool.py

@@ -3,14 +3,12 @@ from typing import Any
 
 from pydantic import BaseModel
 
-from core.plugin.entities.plugin_daemon import (
-    PluginBasicBooleanResponse,
-    PluginToolProviderEntity,
-)
+# from core.plugin.entities.plugin import GenericProviderID, ToolProviderID
+from core.plugin.entities.plugin_daemon import CredentialType, PluginBasicBooleanResponse, PluginToolProviderEntity
 from core.plugin.impl.base import BasePluginClient
 from core.plugin.utils.chunk_merger import merge_blob_chunks
 from core.schemas.resolver import resolve_dify_schema_refs
-from core.tools.entities.tool_entities import CredentialType, ToolInvokeMessage, ToolParameter
+from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 from models.provider_ids import GenericProviderID, ToolProviderID
 
 

api/core/plugin/impl/trigger.py

@@ -0,0 +1,311 @@
+import binascii
+from collections.abc import Generator, Mapping
+from typing import Any
+
+from flask import Request
+
+from core.plugin.entities.plugin_daemon import CredentialType, PluginTriggerProviderEntity
+from core.plugin.entities.request import (
+    PluginTriggerDispatchResponse,
+    TriggerDispatchResponse,
+    TriggerInvokeEventResponse,
+    TriggerSubscriptionResponse,
+    TriggerValidateProviderCredentialsResponse,
+)
+from core.plugin.impl.base import BasePluginClient
+from core.plugin.utils.http_parser import deserialize_response, serialize_request
+from core.trigger.entities.entities import Subscription
+from models.provider_ids import TriggerProviderID
+
+
+class PluginTriggerManager(BasePluginClient):
+    def fetch_trigger_providers(self, tenant_id: str) -> list[PluginTriggerProviderEntity]:
+        """
+        Fetch trigger providers for the given tenant.
+        """
+
+        def transformer(json_response: dict[str, Any]) -> dict:
+            for provider in json_response.get("data", []):
+                declaration = provider.get("declaration", {}) or {}
+                provider_id = provider.get("plugin_id") + "/" + provider.get("provider")
+                for event in declaration.get("events", []):
+                    event["identity"]["provider"] = provider_id
+
+            return json_response
+
+        response: list[PluginTriggerProviderEntity] = self._request_with_plugin_daemon_response(
+            method="GET",
+            path=f"plugin/{tenant_id}/management/triggers",
+            type_=list[PluginTriggerProviderEntity],
+            params={"page": 1, "page_size": 256},
+            transformer=transformer,
+        )
+
+        for provider in response:
+            provider.declaration.identity.name = f"{provider.plugin_id}/{provider.declaration.identity.name}"
+
+            # override the provider name for each trigger to plugin_id/provider_name
+            for event in provider.declaration.events:
+                event.identity.provider = provider.declaration.identity.name
+
+        return response
+
+    def fetch_trigger_provider(self, tenant_id: str, provider_id: TriggerProviderID) -> PluginTriggerProviderEntity:
+        """
+        Fetch trigger provider for the given tenant and plugin.
+        """
+
+        def transformer(json_response: dict[str, Any]) -> dict:
+            data = json_response.get("data")
+            if data:
+                for event in data.get("declaration", {}).get("events", []):
+                    event["identity"]["provider"] = str(provider_id)
+
+            return json_response
+
+        response: PluginTriggerProviderEntity = self._request_with_plugin_daemon_response(
+            method="GET",
+            path=f"plugin/{tenant_id}/management/trigger",
+            type_=PluginTriggerProviderEntity,
+            params={"provider": provider_id.provider_name, "plugin_id": provider_id.plugin_id},
+            transformer=transformer,
+        )
+
+        response.declaration.identity.name = str(provider_id)
+
+        # override the provider name for each trigger to plugin_id/provider_name
+        for event in response.declaration.events:
+            event.identity.provider = str(provider_id)
+
+        return response
+
+    def invoke_trigger_event(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        event_name: str,
+        credentials: Mapping[str, str],
+        credential_type: CredentialType,
+        request: Request,
+        parameters: Mapping[str, Any],
+        subscription: Subscription,
+    ) -> TriggerInvokeEventResponse:
+        """
+        Invoke a trigger with the given parameters.
+        """
+        provider_id = TriggerProviderID(provider)
+        response: Generator[TriggerInvokeEventResponse, None, None] = self._request_with_plugin_daemon_response_stream(
+            method="POST",
+            path=f"plugin/{tenant_id}/dispatch/trigger/invoke_event",
+            type_=TriggerInvokeEventResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider_id.provider_name,
+                    "event": event_name,
+                    "credentials": credentials,
+                    "credential_type": credential_type,
+                    "subscription": subscription.model_dump(),
+                    "raw_http_request": binascii.hexlify(serialize_request(request)).decode(),
+                    "parameters": parameters,
+                },
+            },
+            headers={
+                "X-Plugin-ID": provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp
+
+        raise ValueError("No response received from plugin daemon for invoke trigger")
+
+    def validate_provider_credentials(
+        self, tenant_id: str, user_id: str, provider: str, credentials: Mapping[str, str]
+    ) -> bool:
+        """
+        Validate the credentials of the trigger provider.
+        """
+        provider_id = TriggerProviderID(provider)
+        response: Generator[TriggerValidateProviderCredentialsResponse, None, None] = (
+            self._request_with_plugin_daemon_response_stream(
+                method="POST",
+                path=f"plugin/{tenant_id}/dispatch/trigger/validate_credentials",
+                type_=TriggerValidateProviderCredentialsResponse,
+                data={
+                    "user_id": user_id,
+                    "data": {
+                        "provider": provider_id.provider_name,
+                        "credentials": credentials,
+                    },
+                },
+                headers={
+                    "X-Plugin-ID": provider_id.plugin_id,
+                    "Content-Type": "application/json",
+                },
+            )
+        )
+
+        for resp in response:
+            return resp.result
+
+        raise ValueError("No response received from plugin daemon for validate provider credentials")
+
+    def dispatch_event(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        subscription: Mapping[str, Any],
+        request: Request,
+        credentials: Mapping[str, str],
+        credential_type: CredentialType,
+    ) -> TriggerDispatchResponse:
+        """
+        Dispatch an event to triggers.
+        """
+        provider_id = TriggerProviderID(provider)
+        response: Generator[PluginTriggerDispatchResponse, None, None] = (
+            self._request_with_plugin_daemon_response_stream(
+                method="POST",
+                path=f"plugin/{tenant_id}/dispatch/trigger/dispatch_event",
+                type_=PluginTriggerDispatchResponse,
+                data={
+                    "user_id": user_id,
+                    "data": {
+                        "provider": provider_id.provider_name,
+                        "subscription": subscription,
+                        "credentials": credentials,
+                        "credential_type": credential_type,
+                        "raw_http_request": binascii.hexlify(serialize_request(request)).decode(),
+                    },
+                },
+                headers={
+                    "X-Plugin-ID": provider_id.plugin_id,
+                    "Content-Type": "application/json",
+                },
+            )
+        )
+
+        for resp in response:
+            return TriggerDispatchResponse(
+                events=resp.events,
+                response=deserialize_response(binascii.unhexlify(resp.raw_http_response.encode())),
+            )
+
+        raise ValueError("No response received from plugin daemon for dispatch event")
+
+    def subscribe(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        credentials: Mapping[str, str],
+        credential_type: CredentialType,
+        endpoint: str,
+        parameters: Mapping[str, Any],
+    ) -> TriggerSubscriptionResponse:
+        """
+        Subscribe to a trigger.
+        """
+        provider_id = TriggerProviderID(provider)
+        response: Generator[TriggerSubscriptionResponse, None, None] = self._request_with_plugin_daemon_response_stream(
+            method="POST",
+            path=f"plugin/{tenant_id}/dispatch/trigger/subscribe",
+            type_=TriggerSubscriptionResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider_id.provider_name,
+                    "credentials": credentials,
+                    "credential_type": credential_type,
+                    "endpoint": endpoint,
+                    "parameters": parameters,
+                },
+            },
+            headers={
+                "X-Plugin-ID": provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp
+
+        raise ValueError("No response received from plugin daemon for subscribe")
+
+    def unsubscribe(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        subscription: Subscription,
+        credentials: Mapping[str, str],
+        credential_type: CredentialType,
+    ) -> TriggerSubscriptionResponse:
+        """
+        Unsubscribe from a trigger.
+        """
+        provider_id = TriggerProviderID(provider)
+        response: Generator[TriggerSubscriptionResponse, None, None] = self._request_with_plugin_daemon_response_stream(
+            method="POST",
+            path=f"plugin/{tenant_id}/dispatch/trigger/unsubscribe",
+            type_=TriggerSubscriptionResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider_id.provider_name,
+                    "subscription": subscription.model_dump(),
+                    "credentials": credentials,
+                    "credential_type": credential_type,
+                },
+            },
+            headers={
+                "X-Plugin-ID": provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp
+
+        raise ValueError("No response received from plugin daemon for unsubscribe")
+
+    def refresh(
+        self,
+        tenant_id: str,
+        user_id: str,
+        provider: str,
+        subscription: Subscription,
+        credentials: Mapping[str, str],
+        credential_type: CredentialType,
+    ) -> TriggerSubscriptionResponse:
+        """
+        Refresh a trigger subscription.
+        """
+        provider_id = TriggerProviderID(provider)
+        response: Generator[TriggerSubscriptionResponse, None, None] = self._request_with_plugin_daemon_response_stream(
+            method="POST",
+            path=f"plugin/{tenant_id}/dispatch/trigger/refresh",
+            type_=TriggerSubscriptionResponse,
+            data={
+                "user_id": user_id,
+                "data": {
+                    "provider": provider_id.provider_name,
+                    "subscription": subscription.model_dump(),
+                    "credentials": credentials,
+                    "credential_type": credential_type,
+                },
+            },
+            headers={
+                "X-Plugin-ID": provider_id.plugin_id,
+                "Content-Type": "application/json",
+            },
+        )
+
+        for resp in response:
+            return resp
+
+        raise ValueError("No response received from plugin daemon for refresh")

api/core/plugin/utils/http_parser.py

@@ -0,0 +1,163 @@
+from io import BytesIO
+
+from flask import Request, Response
+from werkzeug.datastructures import Headers
+
+
+def serialize_request(request: Request) -> bytes:
+    method = request.method
+    path = request.full_path.rstrip("?")
+    raw = f"{method} {path} HTTP/1.1\r\n".encode()
+
+    for name, value in request.headers.items():
+        raw += f"{name}: {value}\r\n".encode()
+
+    raw += b"\r\n"
+
+    body = request.get_data(as_text=False)
+    if body:
+        raw += body
+
+    return raw
+
+
+def deserialize_request(raw_data: bytes) -> Request:
+    header_end = raw_data.find(b"\r\n\r\n")
+    if header_end == -1:
+        header_end = raw_data.find(b"\n\n")
+        if header_end == -1:
+            header_data = raw_data
+            body = b""
+        else:
+            header_data = raw_data[:header_end]
+            body = raw_data[header_end + 2 :]
+    else:
+        header_data = raw_data[:header_end]
+        body = raw_data[header_end + 4 :]
+
+    lines = header_data.split(b"\r\n")
+    if len(lines) == 1 and b"\n" in lines[0]:
+        lines = header_data.split(b"\n")
+
+    if not lines or not lines[0]:
+        raise ValueError("Empty HTTP request")
+
+    request_line = lines[0].decode("utf-8", errors="ignore")
+    parts = request_line.split(" ", 2)
+    if len(parts) < 2:
+        raise ValueError(f"Invalid request line: {request_line}")
+
+    method = parts[0]
+    full_path = parts[1]
+    protocol = parts[2] if len(parts) > 2 else "HTTP/1.1"
+
+    if "?" in full_path:
+        path, query_string = full_path.split("?", 1)
+    else:
+        path = full_path
+        query_string = ""
+
+    headers = Headers()
+    for line in lines[1:]:
+        if not line:
+            continue
+        line_str = line.decode("utf-8", errors="ignore")
+        if ":" not in line_str:
+            continue
+        name, value = line_str.split(":", 1)
+        headers.add(name, value.strip())
+
+    host = headers.get("Host", "localhost")
+    if ":" in host:
+        server_name, server_port = host.rsplit(":", 1)
+    else:
+        server_name = host
+        server_port = "80"
+
+    environ = {
+        "REQUEST_METHOD": method,
+        "PATH_INFO": path,
+        "QUERY_STRING": query_string,
+        "SERVER_NAME": server_name,
+        "SERVER_PORT": server_port,
+        "SERVER_PROTOCOL": protocol,
+        "wsgi.input": BytesIO(body),
+        "wsgi.url_scheme": "http",
+    }
+
+    if "Content-Type" in headers:
+        content_type = headers.get("Content-Type")
+        if content_type is not None:
+            environ["CONTENT_TYPE"] = content_type
+
+    if "Content-Length" in headers:
+        content_length = headers.get("Content-Length")
+        if content_length is not None:
+            environ["CONTENT_LENGTH"] = content_length
+    elif body:
+        environ["CONTENT_LENGTH"] = str(len(body))
+
+    for name, value in headers.items():
+        if name.upper() in ("CONTENT-TYPE", "CONTENT-LENGTH"):
+            continue
+        env_name = f"HTTP_{name.upper().replace('-', '_')}"
+        environ[env_name] = value
+
+    return Request(environ)
+
+
+def serialize_response(response: Response) -> bytes:
+    raw = f"HTTP/1.1 {response.status}\r\n".encode()
+
+    for name, value in response.headers.items():
+        raw += f"{name}: {value}\r\n".encode()
+
+    raw += b"\r\n"
+
+    body = response.get_data(as_text=False)
+    if body:
+        raw += body
+
+    return raw
+
+
+def deserialize_response(raw_data: bytes) -> Response:
+    header_end = raw_data.find(b"\r\n\r\n")
+    if header_end == -1:
+        header_end = raw_data.find(b"\n\n")
+        if header_end == -1:
+            header_data = raw_data
+            body = b""
+        else:
+            header_data = raw_data[:header_end]
+            body = raw_data[header_end + 2 :]
+    else:
+        header_data = raw_data[:header_end]
+        body = raw_data[header_end + 4 :]
+
+    lines = header_data.split(b"\r\n")
+    if len(lines) == 1 and b"\n" in lines[0]:
+        lines = header_data.split(b"\n")
+
+    if not lines or not lines[0]:
+        raise ValueError("Empty HTTP response")
+
+    status_line = lines[0].decode("utf-8", errors="ignore")
+    parts = status_line.split(" ", 2)
+    if len(parts) < 2:
+        raise ValueError(f"Invalid status line: {status_line}")
+
+    status_code = int(parts[1])
+
+    response = Response(response=body, status=status_code)
+
+    for line in lines[1:]:
+        if not line:
+            continue
+        line_str = line.decode("utf-8", errors="ignore")
+        if ":" not in line_str:
+            continue
+        name, value = line_str.split(":", 1)
+        response.headers[name] = value.strip()
+
+    return response

api/core/provider_manager.py

@@ -610,7 +610,7 @@ class ProviderManager:
 
             provider_quota_to_provider_record_dict = {}
             for provider_record in provider_records:
-                if provider_record.provider_type != ProviderType.SYSTEM.value:
+                if provider_record.provider_type != ProviderType.SYSTEM:
                     continue
 
                 provider_quota_to_provider_record_dict[ProviderQuotaType.value_of(provider_record.quota_type)] = (
@@ -627,8 +627,8 @@ class ProviderManager:
                                 tenant_id=tenant_id,
                                 # TODO: Use provider name with prefix after the data migration.
                                 provider_name=ModelProviderID(provider_name).provider_name,
-                                provider_type=ProviderType.SYSTEM.value,
-                                quota_type=ProviderQuotaType.TRIAL.value,
+                                provider_type=ProviderType.SYSTEM,
+                                quota_type=ProviderQuotaType.TRIAL,
                                 quota_limit=quota.quota_limit,  # type: ignore
                                 quota_used=0,
                                 is_valid=True,
@@ -641,8 +641,8 @@ class ProviderManager:
                             stmt = select(Provider).where(
                                 Provider.tenant_id == tenant_id,
                                 Provider.provider_name == ModelProviderID(provider_name).provider_name,
-                                Provider.provider_type == ProviderType.SYSTEM.value,
-                                Provider.quota_type == ProviderQuotaType.TRIAL.value,
+                                Provider.provider_type == ProviderType.SYSTEM,
+                                Provider.quota_type == ProviderQuotaType.TRIAL,
                             )
                             existed_provider_record = db.session.scalar(stmt)
                             if not existed_provider_record:
@@ -702,7 +702,7 @@ class ProviderManager:
         """Get custom provider configuration."""
         # Find custom provider record (non-system)
         custom_provider_record = next(
-            (record for record in provider_records if record.provider_type != ProviderType.SYSTEM.value), None
+            (record for record in provider_records if record.provider_type != ProviderType.SYSTEM), None
         )
 
         if not custom_provider_record:
@@ -905,7 +905,7 @@ class ProviderManager:
         # Convert provider_records to dict
         quota_type_to_provider_records_dict: dict[ProviderQuotaType, Provider] = {}
         for provider_record in provider_records:
-            if provider_record.provider_type != ProviderType.SYSTEM.value:
+            if provider_record.provider_type != ProviderType.SYSTEM:
                 continue
 
             quota_type_to_provider_records_dict[ProviderQuotaType.value_of(provider_record.quota_type)] = (
@@ -1046,7 +1046,7 @@ class ProviderManager:
         """
         secret_input_form_variables = []
         for credential_form_schema in credential_form_schemas:
-            if credential_form_schema.type.value == FormType.SECRET_INPUT.value:
+            if credential_form_schema.type == FormType.SECRET_INPUT:
                 secret_input_form_variables.append(credential_form_schema.variable)
 
         return secret_input_form_variables

api/core/rag/data_post_processor/data_post_processor.py

@@ -46,7 +46,7 @@ class DataPostProcessor:
         reranking_model: dict | None = None,
         weights: dict | None = None,
     ) -> BaseRerankRunner | None:
-        if reranking_mode == RerankMode.WEIGHTED_SCORE.value and weights:
+        if reranking_mode == RerankMode.WEIGHTED_SCORE and weights:
             runner = RerankRunnerFactory.create_rerank_runner(
                 runner_type=reranking_mode,
                 tenant_id=tenant_id,
@@ -62,7 +62,7 @@ class DataPostProcessor:
                 ),
             )
             return runner
-        elif reranking_mode == RerankMode.RERANKING_MODEL.value:
+        elif reranking_mode == RerankMode.RERANKING_MODEL:
             rerank_model_instance = self._get_rerank_model_instance(tenant_id, reranking_model)
             if rerank_model_instance is None:
                 return None

api/core/rag/datasource/retrieval_service.py

@@ -21,7 +21,7 @@ from models.dataset import Document as DatasetDocument
 from services.external_knowledge_service import ExternalDatasetService
 
 default_retrieval_model = {
-    "search_method": RetrievalMethod.SEMANTIC_SEARCH.value,
+    "search_method": RetrievalMethod.SEMANTIC_SEARCH,
     "reranking_enable": False,
     "reranking_model": {"reranking_provider_name": "", "reranking_model_name": ""},
     "top_k": 4,
@@ -34,7 +34,7 @@ class RetrievalService:
     @classmethod
     def retrieve(
         cls,
-        retrieval_method: str,
+        retrieval_method: RetrievalMethod,
         dataset_id: str,
         query: str,
         top_k: int,
@@ -56,7 +56,7 @@ class RetrievalService:
         # Optimize multithreading with thread pools
         with ThreadPoolExecutor(max_workers=dify_config.RETRIEVAL_SERVICE_EXECUTORS) as executor:  # type: ignore
             futures = []
-            if retrieval_method == "keyword_search":
+            if retrieval_method == RetrievalMethod.KEYWORD_SEARCH:
                 futures.append(
                     executor.submit(
                         cls.keyword_search,
@@ -107,7 +107,7 @@ class RetrievalService:
             raise ValueError(";\n".join(exceptions))
 
         # Deduplicate documents for hybrid search to avoid duplicate chunks
-        if retrieval_method == RetrievalMethod.HYBRID_SEARCH.value:
+        if retrieval_method == RetrievalMethod.HYBRID_SEARCH:
             all_documents = cls._deduplicate_documents(all_documents)
             data_post_processor = DataPostProcessor(
                 str(dataset.tenant_id), reranking_mode, reranking_model, weights, False
@@ -220,7 +220,7 @@ class RetrievalService:
         score_threshold: float | None,
         reranking_model: dict | None,
         all_documents: list,
-        retrieval_method: str,
+        retrieval_method: RetrievalMethod,
         exceptions: list,
         document_ids_filter: list[str] | None = None,
     ):
@@ -245,10 +245,10 @@ class RetrievalService:
                         reranking_model
                         and reranking_model.get("reranking_model_name")
                         and reranking_model.get("reranking_provider_name")
-                        and retrieval_method == RetrievalMethod.SEMANTIC_SEARCH.value
+                        and retrieval_method == RetrievalMethod.SEMANTIC_SEARCH
                     ):
                         data_post_processor = DataPostProcessor(
-                            str(dataset.tenant_id), str(RerankMode.RERANKING_MODEL.value), reranking_model, None, False
+                            str(dataset.tenant_id), str(RerankMode.RERANKING_MODEL), reranking_model, None, False
                         )
                         all_documents.extend(
                             data_post_processor.invoke(
@@ -293,10 +293,10 @@ class RetrievalService:
                         reranking_model
                         and reranking_model.get("reranking_model_name")
                         and reranking_model.get("reranking_provider_name")
-                        and retrieval_method == RetrievalMethod.FULL_TEXT_SEARCH.value
+                        and retrieval_method == RetrievalMethod.FULL_TEXT_SEARCH
                     ):
                         data_post_processor = DataPostProcessor(
-                            str(dataset.tenant_id), str(RerankMode.RERANKING_MODEL.value), reranking_model, None, False
+                            str(dataset.tenant_id), str(RerankMode.RERANKING_MODEL), reranking_model, None, False
                         )
                         all_documents.extend(
                             data_post_processor.invoke(

api/core/rag/datasource/vdb/alibabacloud_mysql/alibabacloud_mysql_vector.py

@@ -0,0 +1,388 @@
+import hashlib
+import json
+import logging
+import uuid
+from contextlib import contextmanager
+from typing import Any, Literal, cast
+
+import mysql.connector
+from mysql.connector import Error as MySQLError
+from pydantic import BaseModel, model_validator
+
+from configs import dify_config
+from core.rag.datasource.vdb.vector_base import BaseVector
+from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
+from core.rag.datasource.vdb.vector_type import VectorType
+from core.rag.embedding.embedding_base import Embeddings
+from core.rag.models.document import Document
+from extensions.ext_redis import redis_client
+from models.dataset import Dataset
+
+logger = logging.getLogger(__name__)
+
+
+class AlibabaCloudMySQLVectorConfig(BaseModel):
+    host: str
+    port: int
+    user: str
+    password: str
+    database: str
+    max_connection: int
+    charset: str = "utf8mb4"
+    distance_function: Literal["cosine", "euclidean"] = "cosine"
+    hnsw_m: int = 6
+
+    @model_validator(mode="before")
+    @classmethod
+    def validate_config(cls, values: dict):
+        if not values.get("host"):
+            raise ValueError("config ALIBABACLOUD_MYSQL_HOST is required")
+        if not values.get("port"):
+            raise ValueError("config ALIBABACLOUD_MYSQL_PORT is required")
+        if not values.get("user"):
+            raise ValueError("config ALIBABACLOUD_MYSQL_USER is required")
+        if values.get("password") is None:
+            raise ValueError("config ALIBABACLOUD_MYSQL_PASSWORD is required")
+        if not values.get("database"):
+            raise ValueError("config ALIBABACLOUD_MYSQL_DATABASE is required")
+        if not values.get("max_connection"):
+            raise ValueError("config ALIBABACLOUD_MYSQL_MAX_CONNECTION is required")
+        return values
+
+
+SQL_CREATE_TABLE = """
+CREATE TABLE IF NOT EXISTS {table_name} (
+    id VARCHAR(36) PRIMARY KEY,
+    text LONGTEXT NOT NULL,
+    meta JSON NOT NULL,
+    embedding VECTOR({dimension}) NOT NULL,
+    VECTOR INDEX (embedding) M={hnsw_m} DISTANCE={distance_function}
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+"""
+
+SQL_CREATE_META_INDEX = """
+CREATE INDEX idx_{index_hash}_meta ON {table_name}
+    ((CAST(JSON_UNQUOTE(JSON_EXTRACT(meta, '$.document_id')) AS CHAR(36))));
+"""
+
+SQL_CREATE_FULLTEXT_INDEX = """
+CREATE FULLTEXT INDEX idx_{index_hash}_text ON {table_name} (text) WITH PARSER ngram;
+"""
+
+
+class AlibabaCloudMySQLVector(BaseVector):
+    def __init__(self, collection_name: str, config: AlibabaCloudMySQLVectorConfig):
+        super().__init__(collection_name)
+        self.pool = self._create_connection_pool(config)
+        self.table_name = collection_name.lower()
+        self.index_hash = hashlib.md5(self.table_name.encode()).hexdigest()[:8]
+        self.distance_function = config.distance_function.lower()
+        self.hnsw_m = config.hnsw_m
+        self._check_vector_support()
+
+    def get_type(self) -> str:
+        return VectorType.ALIBABACLOUD_MYSQL
+
+    def _create_connection_pool(self, config: AlibabaCloudMySQLVectorConfig):
+        # Create connection pool using mysql-connector-python pooling
+        pool_config: dict[str, Any] = {
+            "host": config.host,
+            "port": config.port,
+            "user": config.user,
+            "password": config.password,
+            "database": config.database,
+            "charset": config.charset,
+            "autocommit": True,
+            "pool_name": f"pool_{self.collection_name}",
+            "pool_size": config.max_connection,
+            "pool_reset_session": True,
+        }
+        return mysql.connector.pooling.MySQLConnectionPool(**pool_config)
+
+    def _check_vector_support(self):
+        """Check if the MySQL server supports vector operations."""
+        try:
+            with self._get_cursor() as cur:
+                # Check MySQL version and vector support
+                cur.execute("SELECT VERSION()")
+                version = cur.fetchone()["VERSION()"]
+                logger.debug("Connected to MySQL version: %s", version)
+                # Try to execute a simple vector function to verify support
+                cur.execute("SELECT VEC_FromText('[1,2,3]') IS NOT NULL as vector_support")
+                result = cur.fetchone()
+                if not result or not result.get("vector_support"):
+                    raise ValueError(
+                        "RDS MySQL Vector functions are not available."
+                        " Please ensure you're using RDS MySQL 8.0.36+ with Vector support."
+                    )
+
+        except MySQLError as e:
+            if "FUNCTION" in str(e) and "VEC_FromText" in str(e):
+                raise ValueError(
+                    "RDS MySQL Vector functions are not available."
+                    " Please ensure you're using RDS MySQL 8.0.36+ with Vector support."
+                ) from e
+            raise e
+
+    @contextmanager
+    def _get_cursor(self):
+        conn = self.pool.get_connection()
+        cur = conn.cursor(dictionary=True)
+        try:
+            yield cur
+        finally:
+            cur.close()
+            conn.close()
+
+    def create(self, texts: list[Document], embeddings: list[list[float]], **kwargs):
+        dimension = len(embeddings[0])
+        self._create_collection(dimension)
+        return self.add_texts(texts, embeddings)
+
+    def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
+        values = []
+        pks = []
+        for i, doc in enumerate(documents):
+            if doc.metadata is not None:
+                doc_id = doc.metadata.get("doc_id", str(uuid.uuid4()))
+                pks.append(doc_id)
+                # Convert embedding list to Aliyun MySQL vector format
+                vector_str = "[" + ",".join(map(str, embeddings[i])) + "]"
+                values.append(
+                    (
+                        doc_id,
+                        doc.page_content,
+                        json.dumps(doc.metadata),
+                        vector_str,
+                    )
+                )
+
+        with self._get_cursor() as cur:
+            insert_sql = (
+                f"INSERT INTO {self.table_name} (id, text, meta, embedding) VALUES (%s, %s, %s, VEC_FromText(%s))"
+            )
+            cur.executemany(insert_sql, values)
+        return pks
+
+    def text_exists(self, id: str) -> bool:
+        with self._get_cursor() as cur:
+            cur.execute(f"SELECT id FROM {self.table_name} WHERE id = %s", (id,))
+            return cur.fetchone() is not None
+
+    def get_by_ids(self, ids: list[str]) -> list[Document]:
+        if not ids:
+            return []
+
+        with self._get_cursor() as cur:
+            placeholders = ",".join(["%s"] * len(ids))
+            cur.execute(f"SELECT meta, text FROM {self.table_name} WHERE id IN ({placeholders})", ids)
+            docs = []
+            for record in cur:
+                metadata = record["meta"]
+                if isinstance(metadata, str):
+                    metadata = json.loads(metadata)
+                docs.append(Document(page_content=record["text"], metadata=metadata))
+        return docs
+
+    def delete_by_ids(self, ids: list[str]):
+        # Avoiding crashes caused by performing delete operations on empty lists
+        if not ids:
+            return
+
+        with self._get_cursor() as cur:
+            try:
+                placeholders = ",".join(["%s"] * len(ids))
+                cur.execute(f"DELETE FROM {self.table_name} WHERE id IN ({placeholders})", ids)
+            except MySQLError as e:
+                if e.errno == 1146:  # Table doesn't exist
+                    logger.warning("Table %s not found, skipping delete operation.", self.table_name)
+                    return
+                else:
+                    raise e
+
+    def delete_by_metadata_field(self, key: str, value: str):
+        with self._get_cursor() as cur:
+            cur.execute(
+                f"DELETE FROM {self.table_name} WHERE JSON_UNQUOTE(JSON_EXTRACT(meta, %s)) = %s", (f"$.{key}", value)
+            )
+
+    def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:
+        """
+        Search the nearest neighbors to a vector using RDS MySQL vector distance functions.
+
+        :param query_vector: The input vector to search for similar items.
+        :return: List of Documents that are nearest to the query vector.
+        """
+        top_k = kwargs.get("top_k", 4)
+        if not isinstance(top_k, int) or top_k <= 0:
+            raise ValueError("top_k must be a positive integer")
+
+        document_ids_filter = kwargs.get("document_ids_filter")
+        where_clause = ""
+        params = []
+
+        if document_ids_filter:
+            placeholders = ",".join(["%s"] * len(document_ids_filter))
+            where_clause = f" WHERE JSON_UNQUOTE(JSON_EXTRACT(meta, '$.document_id')) IN ({placeholders}) "
+            params.extend(document_ids_filter)
+
+        # Convert query vector to RDS MySQL vector format
+        query_vector_str = "[" + ",".join(map(str, query_vector)) + "]"
+
+        # Use RSD MySQL's native vector distance functions
+        with self._get_cursor() as cur:
+            # Choose distance function based on configuration
+            distance_func = "VEC_DISTANCE_COSINE" if self.distance_function == "cosine" else "VEC_DISTANCE_EUCLIDEAN"
+
+            # Note: RDS MySQL optimizer will use vector index when ORDER BY + LIMIT are present
+            # Use column alias in ORDER BY to avoid calculating distance twice
+            sql = f"""
+            SELECT meta, text,
+                   {distance_func}(embedding, VEC_FromText(%s)) AS distance
+            FROM {self.table_name}
+            {where_clause}
+            ORDER BY distance
+            LIMIT %s
+            """
+            query_params = [query_vector_str] + params + [top_k]
+
+            cur.execute(sql, query_params)
+
+            docs = []
+            score_threshold = float(kwargs.get("score_threshold") or 0.0)
+
+            for record in cur:
+                try:
+                    distance = float(record["distance"])
+                    # Convert distance to similarity score
+                    if self.distance_function == "cosine":
+                        # For cosine distance: similarity = 1 - distance
+                        similarity = 1.0 - distance
+                    else:
+                        # For euclidean distance: use inverse relationship
+                        # similarity = 1 / (1 + distance)
+                        similarity = 1.0 / (1.0 + distance)
+
+                    metadata = record["meta"]
+                    if isinstance(metadata, str):
+                        metadata = json.loads(metadata)
+                    metadata["score"] = similarity
+                    metadata["distance"] = distance
+
+                    if similarity >= score_threshold:
+                        docs.append(Document(page_content=record["text"], metadata=metadata))
+                except (ValueError, json.JSONDecodeError) as e:
+                    logger.warning("Error processing search result: %s", e)
+                    continue
+
+            return docs
+
+    def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
+        top_k = kwargs.get("top_k", 5)
+        if not isinstance(top_k, int) or top_k <= 0:
+            raise ValueError("top_k must be a positive integer")
+
+        document_ids_filter = kwargs.get("document_ids_filter")
+        where_clause = ""
+        params = []
+
+        if document_ids_filter:
+            placeholders = ",".join(["%s"] * len(document_ids_filter))
+            where_clause = f" AND JSON_UNQUOTE(JSON_EXTRACT(meta, '$.document_id')) IN ({placeholders}) "
+            params.extend(document_ids_filter)
+
+        with self._get_cursor() as cur:
+            # Build query parameters: query (twice for MATCH clauses), document_ids_filter (if any), top_k
+            query_params = [query, query] + params + [top_k]
+            cur.execute(
+                f"""SELECT meta, text,
+                    MATCH(text) AGAINST(%s IN NATURAL LANGUAGE MODE) AS score
+                    FROM {self.table_name}
+                    WHERE MATCH(text) AGAINST(%s IN NATURAL LANGUAGE MODE)
+                    {where_clause}
+                    ORDER BY score DESC
+                    LIMIT %s""",
+                query_params,
+            )
+            docs = []
+            for record in cur:
+                metadata = record["meta"]
+                if isinstance(metadata, str):
+                    metadata = json.loads(metadata)
+                metadata["score"] = float(record["score"])
+                docs.append(Document(page_content=record["text"], metadata=metadata))
+        return docs
+
+    def delete(self):
+        with self._get_cursor() as cur:
+            cur.execute(f"DROP TABLE IF EXISTS {self.table_name}")
+
+    def _create_collection(self, dimension: int):
+        collection_exist_cache_key = f"vector_indexing_{self._collection_name}"
+        lock_name = f"{collection_exist_cache_key}_lock"
+        with redis_client.lock(lock_name, timeout=20):
+            if redis_client.get(collection_exist_cache_key):
+                return
+
+            with self._get_cursor() as cur:
+                # Create table with vector column and vector index
+                cur.execute(
+                    SQL_CREATE_TABLE.format(
+                        table_name=self.table_name,
+                        dimension=dimension,
+                        distance_function=self.distance_function,
+                        hnsw_m=self.hnsw_m,
+                    )
+                )
+                # Create metadata index (check if exists first)
+                try:
+                    cur.execute(SQL_CREATE_META_INDEX.format(table_name=self.table_name, index_hash=self.index_hash))
+                except MySQLError as e:
+                    if e.errno != 1061:  # Duplicate key name
+                        logger.warning("Could not create meta index: %s", e)
+
+                # Create full-text index for text search
+                try:
+                    cur.execute(
+                        SQL_CREATE_FULLTEXT_INDEX.format(table_name=self.table_name, index_hash=self.index_hash)
+                    )
+                except MySQLError as e:
+                    if e.errno != 1061:  # Duplicate key name
+                        logger.warning("Could not create fulltext index: %s", e)
+
+            redis_client.set(collection_exist_cache_key, 1, ex=3600)
+
+
+class AlibabaCloudMySQLVectorFactory(AbstractVectorFactory):
+    def _validate_distance_function(self, distance_function: str) -> Literal["cosine", "euclidean"]:
+        """Validate and return the distance function as a proper Literal type."""
+        if distance_function not in ["cosine", "euclidean"]:
+            raise ValueError(f"Invalid distance function: {distance_function}. Must be 'cosine' or 'euclidean'")
+        return cast(Literal["cosine", "euclidean"], distance_function)
+
+    def init_vector(self, dataset: Dataset, attributes: list, embeddings: Embeddings) -> AlibabaCloudMySQLVector:
+        if dataset.index_struct_dict:
+            class_prefix: str = dataset.index_struct_dict["vector_store"]["class_prefix"]
+            collection_name = class_prefix
+        else:
+            dataset_id = dataset.id
+            collection_name = Dataset.gen_collection_name_by_id(dataset_id)
+            dataset.index_struct = json.dumps(
+                self.gen_index_struct_dict(VectorType.ALIBABACLOUD_MYSQL, collection_name)
+            )
+        return AlibabaCloudMySQLVector(
+            collection_name=collection_name,
+            config=AlibabaCloudMySQLVectorConfig(
+                host=dify_config.ALIBABACLOUD_MYSQL_HOST or "localhost",
+                port=dify_config.ALIBABACLOUD_MYSQL_PORT,
+                user=dify_config.ALIBABACLOUD_MYSQL_USER or "root",
+                password=dify_config.ALIBABACLOUD_MYSQL_PASSWORD or "",
+                database=dify_config.ALIBABACLOUD_MYSQL_DATABASE or "dify",
+                max_connection=dify_config.ALIBABACLOUD_MYSQL_MAX_CONNECTION,
+                charset=dify_config.ALIBABACLOUD_MYSQL_CHARSET or "utf8mb4",
+                distance_function=self._validate_distance_function(
+                    dify_config.ALIBABACLOUD_MYSQL_DISTANCE_FUNCTION or "cosine"
+                ),
+                hnsw_m=dify_config.ALIBABACLOUD_MYSQL_HNSW_M or 6,
+            ),
+        )

api/core/rag/datasource/vdb/clickzetta/clickzetta_vector.py

@@ -488,9 +488,9 @@ class ClickzettaVector(BaseVector):
         create_table_sql = f"""
         CREATE TABLE IF NOT EXISTS {self._config.schema_name}.{self._table_name} (
             id STRING NOT NULL COMMENT 'Unique document identifier',
-            {Field.CONTENT_KEY.value} STRING NOT NULL COMMENT 'Document text content for search and retrieval',
-            {Field.METADATA_KEY.value} JSON COMMENT 'Document metadata including source, type, and other attributes',
-            {Field.VECTOR.value} VECTOR(FLOAT, {dimension}) NOT NULL COMMENT
+            {Field.CONTENT_KEY} STRING NOT NULL COMMENT 'Document text content for search and retrieval',
+            {Field.METADATA_KEY} JSON COMMENT 'Document metadata including source, type, and other attributes',
+            {Field.VECTOR} VECTOR(FLOAT, {dimension}) NOT NULL COMMENT
                 'High-dimensional embedding vector for semantic similarity search',
             PRIMARY KEY (id)
         ) COMMENT 'Dify RAG knowledge base vector storage table for document embeddings and content'
@@ -519,15 +519,15 @@ class ClickzettaVector(BaseVector):
             existing_indexes = cursor.fetchall()
             for idx in existing_indexes:
                 # Check if vector index already exists on the embedding column
-                if Field.VECTOR.value in str(idx).lower():
-                    logger.info("Vector index already exists on column %s", Field.VECTOR.value)
+                if Field.VECTOR in str(idx).lower():
+                    logger.info("Vector index already exists on column %s", Field.VECTOR)
                     return
         except (RuntimeError, ValueError) as e:
             logger.warning("Failed to check existing indexes: %s", e)
 
         index_sql = f"""
         CREATE VECTOR INDEX IF NOT EXISTS {index_name}
-        ON TABLE {self._config.schema_name}.{self._table_name}({Field.VECTOR.value})
+        ON TABLE {self._config.schema_name}.{self._table_name}({Field.VECTOR})
         PROPERTIES (
             "distance.function" = "{self._config.vector_distance_function}",
             "scalar.type" = "f32",
@@ -560,17 +560,17 @@ class ClickzettaVector(BaseVector):
                 # More precise check: look for inverted index specifically on the content column
                 if (
                     "inverted" in idx_str
-                    and Field.CONTENT_KEY.value.lower() in idx_str
+                    and Field.CONTENT_KEY.lower() in idx_str
                     and (index_name.lower() in idx_str or f"idx_{self._table_name}_text" in idx_str)
                 ):
-                    logger.info("Inverted index already exists on column %s: %s", Field.CONTENT_KEY.value, idx)
+                    logger.info("Inverted index already exists on column %s: %s", Field.CONTENT_KEY, idx)
                     return
         except (RuntimeError, ValueError) as e:
             logger.warning("Failed to check existing indexes: %s", e)
 
         index_sql = f"""
         CREATE INVERTED INDEX IF NOT EXISTS {index_name}
-        ON TABLE {self._config.schema_name}.{self._table_name} ({Field.CONTENT_KEY.value})
+        ON TABLE {self._config.schema_name}.{self._table_name} ({Field.CONTENT_KEY})
         PROPERTIES (
             "analyzer" = "{self._config.analyzer_type}",
             "mode" = "{self._config.analyzer_mode}"
@@ -588,13 +588,13 @@ class ClickzettaVector(BaseVector):
                 or "with the same type" in error_msg
                 or "cannot create inverted index" in error_msg
             ) and "already has index" in error_msg:
-                logger.info("Inverted index already exists on column %s", Field.CONTENT_KEY.value)
+                logger.info("Inverted index already exists on column %s", Field.CONTENT_KEY)
                 # Try to get the existing index name for logging
                 try:
                     cursor.execute(f"SHOW INDEX FROM {self._config.schema_name}.{self._table_name}")
                     existing_indexes = cursor.fetchall()
                     for idx in existing_indexes:
-                        if "inverted" in str(idx).lower() and Field.CONTENT_KEY.value.lower() in str(idx).lower():
+                        if "inverted" in str(idx).lower() and Field.CONTENT_KEY.lower() in str(idx).lower():
                             logger.info("Found existing inverted index: %s", idx)
                             break
                 except (RuntimeError, ValueError):
@@ -669,7 +669,7 @@ class ClickzettaVector(BaseVector):
 
         # Use parameterized INSERT with executemany for better performance and security
         # Cast JSON and VECTOR in SQL, pass raw data as parameters
-        columns = f"id, {Field.CONTENT_KEY.value}, {Field.METADATA_KEY.value}, {Field.VECTOR.value}"
+        columns = f"id, {Field.CONTENT_KEY}, {Field.METADATA_KEY}, {Field.VECTOR}"
         insert_sql = (
             f"INSERT INTO {self._config.schema_name}.{self._table_name} ({columns}) "
             f"VALUES (?, ?, CAST(? AS JSON), CAST(? AS VECTOR({vector_dimension})))"
@@ -767,7 +767,7 @@ class ClickzettaVector(BaseVector):
                 # Use json_extract_string function for ClickZetta compatibility
                 sql = (
                     f"DELETE FROM {self._config.schema_name}.{self._table_name} "
-                    f"WHERE json_extract_string({Field.METADATA_KEY.value}, '$.{key}') = ?"
+                    f"WHERE json_extract_string({Field.METADATA_KEY}, '$.{key}') = ?"
                 )
                 cursor.execute(sql, binding_params=[value])
 
@@ -795,9 +795,7 @@ class ClickzettaVector(BaseVector):
             safe_doc_ids = [str(id).replace("'", "''") for id in document_ids_filter]
             doc_ids_str = ",".join(f"'{id}'" for id in safe_doc_ids)
             # Use json_extract_string function for ClickZetta compatibility
-            filter_clauses.append(
-                f"json_extract_string({Field.METADATA_KEY.value}, '$.document_id') IN ({doc_ids_str})"
-            )
+            filter_clauses.append(f"json_extract_string({Field.METADATA_KEY}, '$.document_id') IN ({doc_ids_str})")
 
         # No need for dataset_id filter since each dataset has its own table
 
@@ -808,23 +806,21 @@ class ClickzettaVector(BaseVector):
             distance_func = "COSINE_DISTANCE"
             if score_threshold > 0:
                 query_vector_str = f"CAST('[{self._format_vector_simple(query_vector)}]' AS VECTOR({vector_dimension}))"
-                filter_clauses.append(
-                    f"{distance_func}({Field.VECTOR.value}, {query_vector_str}) < {2 - score_threshold}"
-                )
+                filter_clauses.append(f"{distance_func}({Field.VECTOR}, {query_vector_str}) < {2 - score_threshold}")
         else:
             # For L2 distance, smaller is better
             distance_func = "L2_DISTANCE"
             if score_threshold > 0:
                 query_vector_str = f"CAST('[{self._format_vector_simple(query_vector)}]' AS VECTOR({vector_dimension}))"
-                filter_clauses.append(f"{distance_func}({Field.VECTOR.value}, {query_vector_str}) < {score_threshold}")
+                filter_clauses.append(f"{distance_func}({Field.VECTOR}, {query_vector_str}) < {score_threshold}")
 
         where_clause = " AND ".join(filter_clauses) if filter_clauses else "1=1"
 
         # Execute vector search query
         query_vector_str = f"CAST('[{self._format_vector_simple(query_vector)}]' AS VECTOR({vector_dimension}))"
         search_sql = f"""
-        SELECT id, {Field.CONTENT_KEY.value}, {Field.METADATA_KEY.value},
-               {distance_func}({Field.VECTOR.value}, {query_vector_str}) AS distance
+        SELECT id, {Field.CONTENT_KEY}, {Field.METADATA_KEY},
+               {distance_func}({Field.VECTOR}, {query_vector_str}) AS distance
         FROM {self._config.schema_name}.{self._table_name}
         WHERE {where_clause}
         ORDER BY distance
@@ -887,9 +883,7 @@ class ClickzettaVector(BaseVector):
             safe_doc_ids = [str(id).replace("'", "''") for id in document_ids_filter]
             doc_ids_str = ",".join(f"'{id}'" for id in safe_doc_ids)
             # Use json_extract_string function for ClickZetta compatibility
-            filter_clauses.append(
-                f"json_extract_string({Field.METADATA_KEY.value}, '$.document_id') IN ({doc_ids_str})"
-            )
+            filter_clauses.append(f"json_extract_string({Field.METADATA_KEY}, '$.document_id') IN ({doc_ids_str})")
 
         # No need for dataset_id filter since each dataset has its own table
 
@@ -897,13 +891,13 @@ class ClickzettaVector(BaseVector):
         # match_all requires all terms to be present
         # Use simple quote escaping for MATCH_ALL since it needs to be in the WHERE clause
         escaped_query = query.replace("'", "''")
-        filter_clauses.append(f"MATCH_ALL({Field.CONTENT_KEY.value}, '{escaped_query}')")
+        filter_clauses.append(f"MATCH_ALL({Field.CONTENT_KEY}, '{escaped_query}')")
 
         where_clause = " AND ".join(filter_clauses)
 
         # Execute full-text search query
         search_sql = f"""
-        SELECT id, {Field.CONTENT_KEY.value}, {Field.METADATA_KEY.value}
+        SELECT id, {Field.CONTENT_KEY}, {Field.METADATA_KEY}
         FROM {self._config.schema_name}.{self._table_name}
         WHERE {where_clause}
         LIMIT {top_k}
@@ -986,19 +980,17 @@ class ClickzettaVector(BaseVector):
             safe_doc_ids = [str(id).replace("'", "''") for id in document_ids_filter]
             doc_ids_str = ",".join(f"'{id}'" for id in safe_doc_ids)
             # Use json_extract_string function for ClickZetta compatibility
-            filter_clauses.append(
-                f"json_extract_string({Field.METADATA_KEY.value}, '$.document_id') IN ({doc_ids_str})"
-            )
+            filter_clauses.append(f"json_extract_string({Field.METADATA_KEY}, '$.document_id') IN ({doc_ids_str})")
 
         # No need for dataset_id filter since each dataset has its own table
 
         # Use simple quote escaping for LIKE clause
         escaped_query = query.replace("'", "''")
-        filter_clauses.append(f"{Field.CONTENT_KEY.value} LIKE '%{escaped_query}%'")
+        filter_clauses.append(f"{Field.CONTENT_KEY} LIKE '%{escaped_query}%'")
         where_clause = " AND ".join(filter_clauses)
 
         search_sql = f"""
-        SELECT id, {Field.CONTENT_KEY.value}, {Field.METADATA_KEY.value}
+        SELECT id, {Field.CONTENT_KEY}, {Field.METADATA_KEY}
         FROM {self._config.schema_name}.{self._table_name}
         WHERE {where_clause}
         LIMIT {top_k}

api/core/rag/datasource/vdb/elasticsearch/elasticsearch_ja_vector.py

@@ -57,18 +57,18 @@ class ElasticSearchJaVector(ElasticSearchVector):
                 }
                 mappings = {
                     "properties": {
-                        Field.CONTENT_KEY.value: {
+                        Field.CONTENT_KEY: {
                             "type": "text",
                             "analyzer": "ja_analyzer",
                             "search_analyzer": "ja_analyzer",
                         },
-                        Field.VECTOR.value: {  # Make sure the dimension is correct here
+                        Field.VECTOR: {  # Make sure the dimension is correct here
                             "type": "dense_vector",
                             "dims": dim,
                             "index": True,
                             "similarity": "cosine",
                         },
-                        Field.METADATA_KEY.value: {
+                        Field.METADATA_KEY: {
                             "type": "object",
                             "properties": {
                                 "doc_id": {"type": "keyword"}  # Map doc_id to keyword type

api/core/rag/datasource/vdb/elasticsearch/elasticsearch_vector.py

@@ -4,7 +4,7 @@ import math
 from typing import Any, cast
 from urllib.parse import urlparse
 
-import requests
+from elasticsearch import ConnectionError as ElasticsearchConnectionError
 from elasticsearch import Elasticsearch
 from flask import current_app
 from packaging.version import parse as parse_version
@@ -138,7 +138,7 @@ class ElasticSearchVector(BaseVector):
             if not client.ping():
                 raise ConnectionError("Failed to connect to Elasticsearch")
 
-        except requests.ConnectionError as e:
+        except ElasticsearchConnectionError as e:
             raise ConnectionError(f"Vector database connection error: {str(e)}")
         except Exception as e:
             raise ConnectionError(f"Elasticsearch client initialization failed: {str(e)}")
@@ -163,9 +163,9 @@ class ElasticSearchVector(BaseVector):
                 index=self._collection_name,
                 id=uuids[i],
                 document={
-                    Field.CONTENT_KEY.value: documents[i].page_content,
-                    Field.VECTOR.value: embeddings[i] or None,
-                    Field.METADATA_KEY.value: documents[i].metadata or {},
+                    Field.CONTENT_KEY: documents[i].page_content,
+                    Field.VECTOR: embeddings[i] or None,
+                    Field.METADATA_KEY: documents[i].metadata or {},
                 },
             )
         self._client.indices.refresh(index=self._collection_name)
@@ -193,7 +193,7 @@ class ElasticSearchVector(BaseVector):
     def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:
         top_k = kwargs.get("top_k", 4)
         num_candidates = math.ceil(top_k * 1.5)
-        knn = {"field": Field.VECTOR.value, "query_vector": query_vector, "k": top_k, "num_candidates": num_candidates}
+        knn = {"field": Field.VECTOR, "query_vector": query_vector, "k": top_k, "num_candidates": num_candidates}
         document_ids_filter = kwargs.get("document_ids_filter")
         if document_ids_filter:
             knn["filter"] = {"terms": {"metadata.document_id": document_ids_filter}}
@@ -205,9 +205,9 @@ class ElasticSearchVector(BaseVector):
             docs_and_scores.append(
                 (
                     Document(
-                        page_content=hit["_source"][Field.CONTENT_KEY.value],
-                        vector=hit["_source"][Field.VECTOR.value],
-                        metadata=hit["_source"][Field.METADATA_KEY.value],
+                        page_content=hit["_source"][Field.CONTENT_KEY],
+                        vector=hit["_source"][Field.VECTOR],
+                        metadata=hit["_source"][Field.METADATA_KEY],
                     ),
                     hit["_score"],
                 )
@@ -224,13 +224,13 @@ class ElasticSearchVector(BaseVector):
         return docs
 
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
-        query_str: dict[str, Any] = {"match": {Field.CONTENT_KEY.value: query}}
+        query_str: dict[str, Any] = {"match": {Field.CONTENT_KEY: query}}
         document_ids_filter = kwargs.get("document_ids_filter")
 
         if document_ids_filter:
             query_str = {
                 "bool": {
-                    "must": {"match": {Field.CONTENT_KEY.value: query}},
+                    "must": {"match": {Field.CONTENT_KEY: query}},
                     "filter": {"terms": {"metadata.document_id": document_ids_filter}},
                 }
             }
@@ -240,9 +240,9 @@ class ElasticSearchVector(BaseVector):
         for hit in results["hits"]["hits"]:
             docs.append(
                 Document(
-                    page_content=hit["_source"][Field.CONTENT_KEY.value],
-                    vector=hit["_source"][Field.VECTOR.value],
-                    metadata=hit["_source"][Field.METADATA_KEY.value],
+                    page_content=hit["_source"][Field.CONTENT_KEY],
+                    vector=hit["_source"][Field.VECTOR],
+                    metadata=hit["_source"][Field.METADATA_KEY],
                 )
             )
 
@@ -270,14 +270,14 @@ class ElasticSearchVector(BaseVector):
                 dim = len(embeddings[0])
                 mappings = {
                     "properties": {
-                        Field.CONTENT_KEY.value: {"type": "text"},
-                        Field.VECTOR.value: {  # Make sure the dimension is correct here
+                        Field.CONTENT_KEY: {"type": "text"},
+                        Field.VECTOR: {  # Make sure the dimension is correct here
                             "type": "dense_vector",
                             "dims": dim,
                             "index": True,
                             "similarity": "cosine",
                         },
-                        Field.METADATA_KEY.value: {
+                        Field.METADATA_KEY: {
                             "type": "object",
                             "properties": {
                                 "doc_id": {"type": "keyword"},  # Map doc_id to keyword type

api/core/rag/datasource/vdb/huawei/huawei_cloud_vector.py

@@ -67,9 +67,9 @@ class HuaweiCloudVector(BaseVector):
                 index=self._collection_name,
                 id=uuids[i],
                 document={
-                    Field.CONTENT_KEY.value: documents[i].page_content,
-                    Field.VECTOR.value: embeddings[i] or None,
-                    Field.METADATA_KEY.value: documents[i].metadata or {},
+                    Field.CONTENT_KEY: documents[i].page_content,
+                    Field.VECTOR: embeddings[i] or None,
+                    Field.METADATA_KEY: documents[i].metadata or {},
                 },
             )
         self._client.indices.refresh(index=self._collection_name)
@@ -101,7 +101,7 @@ class HuaweiCloudVector(BaseVector):
             "size": top_k,
             "query": {
                 "vector": {
-                    Field.VECTOR.value: {
+                    Field.VECTOR: {
                         "vector": query_vector,
                         "topk": top_k,
                     }
@@ -116,9 +116,9 @@ class HuaweiCloudVector(BaseVector):
             docs_and_scores.append(
                 (
                     Document(
-                        page_content=hit["_source"][Field.CONTENT_KEY.value],
-                        vector=hit["_source"][Field.VECTOR.value],
-                        metadata=hit["_source"][Field.METADATA_KEY.value],
+                        page_content=hit["_source"][Field.CONTENT_KEY],
+                        vector=hit["_source"][Field.VECTOR],
+                        metadata=hit["_source"][Field.METADATA_KEY],
                     ),
                     hit["_score"],
                 )
@@ -135,15 +135,15 @@ class HuaweiCloudVector(BaseVector):
         return docs
 
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
-        query_str = {"match": {Field.CONTENT_KEY.value: query}}
+        query_str = {"match": {Field.CONTENT_KEY: query}}
         results = self._client.search(index=self._collection_name, query=query_str, size=kwargs.get("top_k", 4))
         docs = []
         for hit in results["hits"]["hits"]:
             docs.append(
                 Document(
-                    page_content=hit["_source"][Field.CONTENT_KEY.value],
-                    vector=hit["_source"][Field.VECTOR.value],
-                    metadata=hit["_source"][Field.METADATA_KEY.value],
+                    page_content=hit["_source"][Field.CONTENT_KEY],
+                    vector=hit["_source"][Field.VECTOR],
+                    metadata=hit["_source"][Field.METADATA_KEY],
                 )
             )
 
@@ -171,8 +171,8 @@ class HuaweiCloudVector(BaseVector):
                 dim = len(embeddings[0])
                 mappings = {
                     "properties": {
-                        Field.CONTENT_KEY.value: {"type": "text"},
-                        Field.VECTOR.value: {  # Make sure the dimension is correct here
+                        Field.CONTENT_KEY: {"type": "text"},
+                        Field.VECTOR: {  # Make sure the dimension is correct here
                             "type": "vector",
                             "dimension": dim,
                             "indexing": True,
@@ -181,7 +181,7 @@ class HuaweiCloudVector(BaseVector):
                             "neighbors": 32,
                             "efc": 128,
                         },
-                        Field.METADATA_KEY.value: {
+                        Field.METADATA_KEY: {
                             "type": "object",
                             "properties": {
                                 "doc_id": {"type": "keyword"}  # Map doc_id to keyword type

api/core/rag/datasource/vdb/lindorm/lindorm_vector.py

@@ -125,9 +125,9 @@ class LindormVectorStore(BaseVector):
                     }
                 }
                 action_values: dict[str, Any] = {
-                    Field.CONTENT_KEY.value: documents[i].page_content,
-                    Field.VECTOR.value: embeddings[i],
-                    Field.METADATA_KEY.value: documents[i].metadata,
+                    Field.CONTENT_KEY: documents[i].page_content,
+                    Field.VECTOR: embeddings[i],
+                    Field.METADATA_KEY: documents[i].metadata,
                 }
                 if self._using_ugc:
                     action_header["index"]["routing"] = self._routing
@@ -149,7 +149,7 @@ class LindormVectorStore(BaseVector):
 
     def get_ids_by_metadata_field(self, key: str, value: str):
         query: dict[str, Any] = {
-            "query": {"bool": {"must": [{"term": {f"{Field.METADATA_KEY.value}.{key}.keyword": value}}]}}
+            "query": {"bool": {"must": [{"term": {f"{Field.METADATA_KEY}.{key}.keyword": value}}]}}
         }
         if self._using_ugc:
             query["query"]["bool"]["must"].append({"term": {f"{ROUTING_FIELD}.keyword": self._routing}})
@@ -252,14 +252,14 @@ class LindormVectorStore(BaseVector):
         search_query: dict[str, Any] = {
             "size": top_k,
             "_source": True,
-            "query": {"knn": {Field.VECTOR.value: {"vector": query_vector, "k": top_k}}},
+            "query": {"knn": {Field.VECTOR: {"vector": query_vector, "k": top_k}}},
         }
 
         final_ext: dict[str, Any] = {"lvector": {}}
         if filters is not None and len(filters) > 0:
             # when using filter, transform filter from List[Dict] to Dict as valid format
             filter_dict = {"bool": {"must": filters}} if len(filters) > 1 else filters[0]
-            search_query["query"]["knn"][Field.VECTOR.value]["filter"] = filter_dict  # filter should be Dict
+            search_query["query"]["knn"][Field.VECTOR]["filter"] = filter_dict  # filter should be Dict
             final_ext["lvector"]["filter_type"] = "pre_filter"
 
         if final_ext != {"lvector": {}}:
@@ -279,9 +279,9 @@ class LindormVectorStore(BaseVector):
             docs_and_scores.append(
                 (
                     Document(
-                        page_content=hit["_source"][Field.CONTENT_KEY.value],
-                        vector=hit["_source"][Field.VECTOR.value],
-                        metadata=hit["_source"][Field.METADATA_KEY.value],
+                        page_content=hit["_source"][Field.CONTENT_KEY],
+                        vector=hit["_source"][Field.VECTOR],
+                        metadata=hit["_source"][Field.METADATA_KEY],
                     ),
                     hit["_score"],
                 )
@@ -318,9 +318,9 @@ class LindormVectorStore(BaseVector):
 
         docs = []
         for hit in response["hits"]["hits"]:
-            metadata = hit["_source"].get(Field.METADATA_KEY.value)
-            vector = hit["_source"].get(Field.VECTOR.value)
-            page_content = hit["_source"].get(Field.CONTENT_KEY.value)
+            metadata = hit["_source"].get(Field.METADATA_KEY)
+            vector = hit["_source"].get(Field.VECTOR)
+            page_content = hit["_source"].get(Field.CONTENT_KEY)
             doc = Document(page_content=page_content, vector=vector, metadata=metadata)
             docs.append(doc)
 
@@ -342,8 +342,8 @@ class LindormVectorStore(BaseVector):
                     "settings": {"index": {"knn": True, "knn_routing": self._using_ugc}},
                     "mappings": {
                         "properties": {
-                            Field.CONTENT_KEY.value: {"type": "text"},
-                            Field.VECTOR.value: {
+                            Field.CONTENT_KEY: {"type": "text"},
+                            Field.VECTOR: {
                                 "type": "knn_vector",
                                 "dimension": len(embeddings[0]),  # Make sure the dimension is correct here
                                 "method": {

api/core/rag/datasource/vdb/milvus/milvus_vector.py

@@ -85,7 +85,7 @@ class MilvusVector(BaseVector):
             collection_info = self._client.describe_collection(self._collection_name)
             fields = [field["name"] for field in collection_info["fields"]]
         # Since primary field is auto-id, no need to track it
-        self._fields = [f for f in fields if f != Field.PRIMARY_KEY.value]
+        self._fields = [f for f in fields if f != Field.PRIMARY_KEY]
 
     def _check_hybrid_search_support(self) -> bool:
         """
@@ -130,9 +130,9 @@ class MilvusVector(BaseVector):
             insert_dict = {
                 # Do not need to insert the sparse_vector field separately, as the text_bm25_emb
                 # function will automatically convert the native text into a sparse vector for us.
-                Field.CONTENT_KEY.value: documents[i].page_content,
-                Field.VECTOR.value: embeddings[i],
-                Field.METADATA_KEY.value: documents[i].metadata,
+                Field.CONTENT_KEY: documents[i].page_content,
+                Field.VECTOR: embeddings[i],
+                Field.METADATA_KEY: documents[i].metadata,
             }
             insert_dict_list.append(insert_dict)
         # Total insert count
@@ -243,15 +243,15 @@ class MilvusVector(BaseVector):
         results = self._client.search(
             collection_name=self._collection_name,
             data=[query_vector],
-            anns_field=Field.VECTOR.value,
+            anns_field=Field.VECTOR,
             limit=kwargs.get("top_k", 4),
-            output_fields=[Field.CONTENT_KEY.value, Field.METADATA_KEY.value],
+            output_fields=[Field.CONTENT_KEY, Field.METADATA_KEY],
             filter=filter,
         )
 
         return self._process_search_results(
             results,
-            output_fields=[Field.CONTENT_KEY.value, Field.METADATA_KEY.value],
+            output_fields=[Field.CONTENT_KEY, Field.METADATA_KEY],
             score_threshold=float(kwargs.get("score_threshold") or 0.0),
         )
 
@@ -264,7 +264,7 @@ class MilvusVector(BaseVector):
                 "Full-text search is disabled: set MILVUS_ENABLE_HYBRID_SEARCH=true (requires Milvus >= 2.5.0)."
             )
             return []
-        if not self.field_exists(Field.SPARSE_VECTOR.value):
+        if not self.field_exists(Field.SPARSE_VECTOR):
             logger.warning(
                 "Full-text search unavailable: collection missing 'sparse_vector' field; "
                 "recreate the collection after enabling MILVUS_ENABLE_HYBRID_SEARCH to add BM25 sparse index."
@@ -279,15 +279,15 @@ class MilvusVector(BaseVector):
         results = self._client.search(
             collection_name=self._collection_name,
             data=[query],
-            anns_field=Field.SPARSE_VECTOR.value,
+            anns_field=Field.SPARSE_VECTOR,
             limit=kwargs.get("top_k", 4),
-            output_fields=[Field.CONTENT_KEY.value, Field.METADATA_KEY.value],
+            output_fields=[Field.CONTENT_KEY, Field.METADATA_KEY],
             filter=filter,
         )
 
         return self._process_search_results(
             results,
-            output_fields=[Field.CONTENT_KEY.value, Field.METADATA_KEY.value],
+            output_fields=[Field.CONTENT_KEY, Field.METADATA_KEY],
             score_threshold=float(kwargs.get("score_threshold") or 0.0),
         )
 
@@ -311,7 +311,7 @@ class MilvusVector(BaseVector):
                 dim = len(embeddings[0])
                 fields = []
                 if metadatas:
-                    fields.append(FieldSchema(Field.METADATA_KEY.value, DataType.JSON, max_length=65_535))
+                    fields.append(FieldSchema(Field.METADATA_KEY, DataType.JSON, max_length=65_535))
 
                 # Create the text field, enable_analyzer will be set True to support milvus automatically
                 # transfer text to sparse_vector, reference: https://milvus.io/docs/full-text-search.md
@@ -326,15 +326,15 @@ class MilvusVector(BaseVector):
                 ):
                     content_field_kwargs["analyzer_params"] = self._client_config.analyzer_params
 
-                fields.append(FieldSchema(Field.CONTENT_KEY.value, DataType.VARCHAR, **content_field_kwargs))
+                fields.append(FieldSchema(Field.CONTENT_KEY, DataType.VARCHAR, **content_field_kwargs))
 
                 # Create the primary key field
-                fields.append(FieldSchema(Field.PRIMARY_KEY.value, DataType.INT64, is_primary=True, auto_id=True))
+                fields.append(FieldSchema(Field.PRIMARY_KEY, DataType.INT64, is_primary=True, auto_id=True))
                 # Create the vector field, supports binary or float vectors
-                fields.append(FieldSchema(Field.VECTOR.value, infer_dtype_bydata(embeddings[0]), dim=dim))
+                fields.append(FieldSchema(Field.VECTOR, infer_dtype_bydata(embeddings[0]), dim=dim))
                 # Create Sparse Vector Index for the collection
                 if self._hybrid_search_enabled:
-                    fields.append(FieldSchema(Field.SPARSE_VECTOR.value, DataType.SPARSE_FLOAT_VECTOR))
+                    fields.append(FieldSchema(Field.SPARSE_VECTOR, DataType.SPARSE_FLOAT_VECTOR))
 
                 schema = CollectionSchema(fields)
 
@@ -342,8 +342,8 @@ class MilvusVector(BaseVector):
                 if self._hybrid_search_enabled:
                     bm25_function = Function(
                         name="text_bm25_emb",
-                        input_field_names=[Field.CONTENT_KEY.value],
-                        output_field_names=[Field.SPARSE_VECTOR.value],
+                        input_field_names=[Field.CONTENT_KEY],
+                        output_field_names=[Field.SPARSE_VECTOR],
                         function_type=FunctionType.BM25,
                     )
                     schema.add_function(bm25_function)
@@ -352,12 +352,12 @@ class MilvusVector(BaseVector):
 
                 # Create Index params for the collection
                 index_params_obj = IndexParams()
-                index_params_obj.add_index(field_name=Field.VECTOR.value, **index_params)
+                index_params_obj.add_index(field_name=Field.VECTOR, **index_params)
 
                 # Create Sparse Vector Index for the collection
                 if self._hybrid_search_enabled:
                     index_params_obj.add_index(
-                        field_name=Field.SPARSE_VECTOR.value, index_type="AUTOINDEX", metric_type="BM25"
+                        field_name=Field.SPARSE_VECTOR, index_type="AUTOINDEX", metric_type="BM25"
                     )
 
                 # Create the collection

api/core/rag/datasource/vdb/opensearch/opensearch_vector.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import Any, Literal
+from typing import Any
 from uuid import uuid4
 
 from opensearchpy import OpenSearch, Urllib3AWSV4SignerAuth, Urllib3HttpConnection, helpers
@@ -8,6 +8,7 @@ from opensearchpy.helpers import BulkIndexError
 from pydantic import BaseModel, model_validator
 
 from configs import dify_config
+from configs.middleware.vdb.opensearch_config import AuthMethod
 from core.rag.datasource.vdb.field import Field
 from core.rag.datasource.vdb.vector_base import BaseVector
 from core.rag.datasource.vdb.vector_factory import AbstractVectorFactory
@@ -25,7 +26,7 @@ class OpenSearchConfig(BaseModel):
     port: int
     secure: bool = False  # use_ssl
     verify_certs: bool = True
-    auth_method: Literal["basic", "aws_managed_iam"] = "basic"
+    auth_method: AuthMethod = AuthMethod.BASIC
     user: str | None = None
     password: str | None = None
     aws_region: str | None = None
@@ -98,9 +99,9 @@ class OpenSearchVector(BaseVector):
                 "_op_type": "index",
                 "_index": self._collection_name.lower(),
                 "_source": {
-                    Field.CONTENT_KEY.value: documents[i].page_content,
-                    Field.VECTOR.value: embeddings[i],  # Make sure you pass an array here
-                    Field.METADATA_KEY.value: documents[i].metadata,
+                    Field.CONTENT_KEY: documents[i].page_content,
+                    Field.VECTOR: embeddings[i],  # Make sure you pass an array here
+                    Field.METADATA_KEY: documents[i].metadata,
                 },
             }
             # See https://github.com/langchain-ai/langchainjs/issues/4346#issuecomment-1935123377
@@ -116,7 +117,7 @@ class OpenSearchVector(BaseVector):
         )
 
     def get_ids_by_metadata_field(self, key: str, value: str):
-        query = {"query": {"term": {f"{Field.METADATA_KEY.value}.{key}": value}}}
+        query = {"query": {"term": {f"{Field.METADATA_KEY}.{key}": value}}}
         response = self._client.search(index=self._collection_name.lower(), body=query)
         if response["hits"]["hits"]:
             return [hit["_id"] for hit in response["hits"]["hits"]]
@@ -180,17 +181,17 @@ class OpenSearchVector(BaseVector):
 
         query = {
             "size": kwargs.get("top_k", 4),
-            "query": {"knn": {Field.VECTOR.value: {Field.VECTOR.value: query_vector, "k": kwargs.get("top_k", 4)}}},
+            "query": {"knn": {Field.VECTOR: {Field.VECTOR: query_vector, "k": kwargs.get("top_k", 4)}}},
         }
         document_ids_filter = kwargs.get("document_ids_filter")
         if document_ids_filter:
             query["query"] = {
                 "script_score": {
-                    "query": {"bool": {"filter": [{"terms": {Field.DOCUMENT_ID.value: document_ids_filter}}]}},
+                    "query": {"bool": {"filter": [{"terms": {Field.DOCUMENT_ID: document_ids_filter}}]}},
                     "script": {
                         "source": "knn_score",
                         "lang": "knn",
-                        "params": {"field": Field.VECTOR.value, "query_value": query_vector, "space_type": "l2"},
+                        "params": {"field": Field.VECTOR, "query_value": query_vector, "space_type": "l2"},
                     },
                 }
             }
@@ -203,7 +204,7 @@ class OpenSearchVector(BaseVector):
 
         docs = []
         for hit in response["hits"]["hits"]:
-            metadata = hit["_source"].get(Field.METADATA_KEY.value, {})
+            metadata = hit["_source"].get(Field.METADATA_KEY, {})
 
             # Make sure metadata is a dictionary
             if metadata is None:
@@ -212,7 +213,7 @@ class OpenSearchVector(BaseVector):
             metadata["score"] = hit["_score"]
             score_threshold = float(kwargs.get("score_threshold") or 0.0)
             if hit["_score"] >= score_threshold:
-                doc = Document(page_content=hit["_source"].get(Field.CONTENT_KEY.value), metadata=metadata)
+                doc = Document(page_content=hit["_source"].get(Field.CONTENT_KEY), metadata=metadata)
                 docs.append(doc)
 
         return docs
@@ -227,9 +228,9 @@ class OpenSearchVector(BaseVector):
 
         docs = []
         for hit in response["hits"]["hits"]:
-            metadata = hit["_source"].get(Field.METADATA_KEY.value)
-            vector = hit["_source"].get(Field.VECTOR.value)
-            page_content = hit["_source"].get(Field.CONTENT_KEY.value)
+            metadata = hit["_source"].get(Field.METADATA_KEY)
+            vector = hit["_source"].get(Field.VECTOR)
+            page_content = hit["_source"].get(Field.CONTENT_KEY)
             doc = Document(page_content=page_content, vector=vector, metadata=metadata)
             docs.append(doc)
 
@@ -250,8 +251,8 @@ class OpenSearchVector(BaseVector):
                     "settings": {"index": {"knn": True}},
                     "mappings": {
                         "properties": {
-                            Field.CONTENT_KEY.value: {"type": "text"},
-                            Field.VECTOR.value: {
+                            Field.CONTENT_KEY: {"type": "text"},
+                            Field.VECTOR: {
                                 "type": "knn_vector",
                                 "dimension": len(embeddings[0]),  # Make sure the dimension is correct here
                                 "method": {
@@ -261,7 +262,7 @@ class OpenSearchVector(BaseVector):
                                     "parameters": {"ef_construction": 64, "m": 8},
                                 },
                             },
-                            Field.METADATA_KEY.value: {
+                            Field.METADATA_KEY: {
                                 "type": "object",
                                 "properties": {
                                     "doc_id": {"type": "keyword"},  # Map doc_id to keyword type
@@ -293,7 +294,7 @@ class OpenSearchVectorFactory(AbstractVectorFactory):
             port=dify_config.OPENSEARCH_PORT,
             secure=dify_config.OPENSEARCH_SECURE,
             verify_certs=dify_config.OPENSEARCH_VERIFY_CERTS,
-            auth_method=dify_config.OPENSEARCH_AUTH_METHOD.value,
+            auth_method=dify_config.OPENSEARCH_AUTH_METHOD,
             user=dify_config.OPENSEARCH_USER,
             password=dify_config.OPENSEARCH_PASSWORD,
             aws_region=dify_config.OPENSEARCH_AWS_REGION,

api/core/rag/datasource/vdb/qdrant/qdrant_vector.py

@@ -147,15 +147,13 @@ class QdrantVector(BaseVector):
 
                 # create group_id payload index
                 self._client.create_payload_index(
-                    collection_name, Field.GROUP_KEY.value, field_schema=PayloadSchemaType.KEYWORD
+                    collection_name, Field.GROUP_KEY, field_schema=PayloadSchemaType.KEYWORD
                 )
                 # create doc_id payload index
-                self._client.create_payload_index(
-                    collection_name, Field.DOC_ID.value, field_schema=PayloadSchemaType.KEYWORD
-                )
+                self._client.create_payload_index(collection_name, Field.DOC_ID, field_schema=PayloadSchemaType.KEYWORD)
                 # create document_id payload index
                 self._client.create_payload_index(
-                    collection_name, Field.DOCUMENT_ID.value, field_schema=PayloadSchemaType.KEYWORD
+                    collection_name, Field.DOCUMENT_ID, field_schema=PayloadSchemaType.KEYWORD
                 )
                 # create full text index
                 text_index_params = TextIndexParams(
@@ -165,9 +163,7 @@ class QdrantVector(BaseVector):
                     max_token_len=20,
                     lowercase=True,
                 )
-                self._client.create_payload_index(
-                    collection_name, Field.CONTENT_KEY.value, field_schema=text_index_params
-                )
+                self._client.create_payload_index(collection_name, Field.CONTENT_KEY, field_schema=text_index_params)
             redis_client.set(collection_exist_cache_key, 1, ex=3600)
 
     def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
@@ -220,10 +216,10 @@ class QdrantVector(BaseVector):
                     self._build_payloads(
                         batch_texts,
                         batch_metadatas,
-                        Field.CONTENT_KEY.value,
-                        Field.METADATA_KEY.value,
+                        Field.CONTENT_KEY,
+                        Field.METADATA_KEY,
                         group_id or "",  # Ensure group_id is never None
-                        Field.GROUP_KEY.value,
+                        Field.GROUP_KEY,
                     ),
                 )
             ]
@@ -381,12 +377,12 @@ class QdrantVector(BaseVector):
         for result in results:
             if result.payload is None:
                 continue
-            metadata = result.payload.get(Field.METADATA_KEY.value) or {}
+            metadata = result.payload.get(Field.METADATA_KEY) or {}
             # duplicate check score threshold
             if result.score >= score_threshold:
                 metadata["score"] = result.score
                 doc = Document(
-                    page_content=result.payload.get(Field.CONTENT_KEY.value, ""),
+                    page_content=result.payload.get(Field.CONTENT_KEY, ""),
                     metadata=metadata,
                 )
                 docs.append(doc)
@@ -433,7 +429,7 @@ class QdrantVector(BaseVector):
         documents = []
         for result in results:
             if result:
-                document = self._document_from_scored_point(result, Field.CONTENT_KEY.value, Field.METADATA_KEY.value)
+                document = self._document_from_scored_point(result, Field.CONTENT_KEY, Field.METADATA_KEY)
                 documents.append(document)
 
         return documents