fix: change default value of expires_at field in tool_builtin_providers to -1

LGTM

.devcontainer/post_create_command.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-npm add -g pnpm@10.11.1
+npm add -g pnpm@10.13.1
 cd web && pnpm install
 pipx install uv
 
@@ -12,3 +12,4 @@ echo 'alias start-containers="cd /workspaces/dify/docker && docker-compose -f do
 echo 'alias stop-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env down"' >> ~/.bashrc
 
 source /home/vscode/.bashrc
+

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -16,6 +16,8 @@ body:
           required: true
         - label: I confirm that I am using English to submit this report, otherwise it will be closed.
           required: true
+        - label: 【中文用户 & Non English User】请使用英语提交，否则会被关闭 ：）
+          required: true
         - label: "Please do not modify this template :) and fill in all the required fields."
           required: true
 

.github/workflows/build-push.yml

@@ -6,6 +6,7 @@ on:
       - "main"
       - "deploy/dev"
       - "deploy/enterprise"
+      - "build/**"
     tags:
       - "*"
 

.github/workflows/style.yml

@@ -28,7 +28,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v45
+        uses: tj-actions/changed-files@v46
         with:
           files: |
             api/**
@@ -75,7 +75,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v45
+        uses: tj-actions/changed-files@v46
         with:
           files: web/**
 
@@ -113,7 +113,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v45
+        uses: tj-actions/changed-files@v46
         with:
           files: |
             docker/generate_docker_compose
@@ -144,7 +144,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v45
+        uses: tj-actions/changed-files@v46
         with:
           files: |
             **.sh
@@ -152,13 +152,15 @@ jobs:
             **.yml
             **Dockerfile
             dev/**
+            .editorconfig
 
       - name: Super-linter
-        uses: super-linter/super-linter/slim@v7
+        uses: super-linter/super-linter/slim@v8
         if: steps.changed-files.outputs.any_changed == 'true'
         env:
           BASH_SEVERITY: warning
-          DEFAULT_BRANCH: main
+          DEFAULT_BRANCH: origin/main
+          EDITORCONFIG_FILE_NAME: editorconfig-checker.json
           FILTER_REGEX_INCLUDE: pnpm-lock.yaml
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           IGNORE_GENERATED_FILES: true
@@ -168,16 +170,6 @@ jobs:
           # FIXME: temporarily disabled until api-docker.yaml's run script is fixed for shellcheck
           # VALIDATE_GITHUB_ACTIONS: true
           VALIDATE_DOCKERFILE_HADOLINT: true
+          VALIDATE_EDITORCONFIG: true
           VALIDATE_XML: true
           VALIDATE_YAML: true
-
-      - name: EditorConfig checks
-        uses: super-linter/super-linter/slim@v7
-        env:
-          DEFAULT_BRANCH: main
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          IGNORE_GENERATED_FILES: true
-          IGNORE_GITIGNORED_FILES: true
-          # EditorConfig validation
-          VALIDATE_EDITORCONFIG: true
-          EDITORCONFIG_FILE_NAME: editorconfig-checker.json

.github/workflows/web-tests.yml

@@ -27,7 +27,7 @@ jobs:
 
       - name: Check changed files
         id: changed-files
-        uses: tj-actions/changed-files@v45
+        uses: tj-actions/changed-files@v46
         with:
           files: web/**
 

api/.env.example

@@ -5,17 +5,17 @@
 SECRET_KEY=
 
 # Console API base URL
-CONSOLE_API_URL=http://127.0.0.1:5001
-CONSOLE_WEB_URL=http://127.0.0.1:3000
+CONSOLE_API_URL=http://localhost:5001
+CONSOLE_WEB_URL=http://localhost:3000
 
 # Service API base URL
-SERVICE_API_URL=http://127.0.0.1:5001
+SERVICE_API_URL=http://localhost:5001
 
 # Web APP base URL
-APP_WEB_URL=http://127.0.0.1:3000
+APP_WEB_URL=http://localhost:3000
 
 # Files URL
-FILES_URL=http://127.0.0.1:5001
+FILES_URL=http://localhost:5001
 
 # INTERNAL_FILES_URL is used for plugin daemon communication within Docker network.
 # Set this to the internal Docker service URL for proper plugin file access.
@@ -54,7 +54,7 @@ REDIS_CLUSTERS_PASSWORD=
 
 # celery configuration
 CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1
-
+CELERY_BACKEND=redis
 # PostgreSQL database configuration
 DB_USERNAME=postgres
 DB_PASSWORD=difyai123456
@@ -138,12 +138,14 @@ SUPABASE_API_KEY=your-access-key
 SUPABASE_URL=your-server-url
 
 # CORS configuration
-WEB_API_CORS_ALLOW_ORIGINS=http://127.0.0.1:3000,*
-CONSOLE_CORS_ALLOW_ORIGINS=http://127.0.0.1:3000,*
+WEB_API_CORS_ALLOW_ORIGINS=http://localhost:3000,*
+CONSOLE_CORS_ALLOW_ORIGINS=http://localhost:3000,*
 
 # Vector database configuration
-# support: weaviate, qdrant, milvus, myscale, relyt, pgvecto_rs, pgvector, pgvector, chroma, opensearch, tidb_vector, couchbase, vikingdb, upstash, lindorm, oceanbase, opengauss, tablestore, matrixone
+# Supported values are `weaviate`, `qdrant`, `milvus`, `myscale`, `relyt`, `pgvector`, `pgvecto-rs`, `chroma`, `opensearch`, `oracle`, `tencent`, `elasticsearch`, `elasticsearch-ja`, `analyticdb`, `couchbase`, `vikingdb`, `oceanbase`, `opengauss`, `tablestore`,`vastbase`,`tidb`,`tidb_on_qdrant`,`baidu`,`lindorm`,`huawei_cloud`,`upstash`, `matrixone`.
 VECTOR_STORE=weaviate
+# Prefix used to create collection name in vector database
+VECTOR_INDEX_NAME_PREFIX=Vector_index
 
 # Weaviate configuration
 WEAVIATE_ENDPOINT=http://localhost:8080
@@ -495,6 +497,8 @@ ENDPOINT_URL_TEMPLATE=http://localhost:5002/e/{hook_id}
 
 # Reset password token expiry minutes
 RESET_PASSWORD_TOKEN_EXPIRY_MINUTES=5
+CHANGE_EMAIL_TOKEN_EXPIRY_MINUTES=5
+OWNER_TRANSFER_TOKEN_EXPIRY_MINUTES=5
 
 CREATE_TIDB_SERVICE_JOB_ENABLED=false
 
@@ -505,6 +509,8 @@ LOGIN_LOCKOUT_DURATION=86400
 
 # Enable OpenTelemetry
 ENABLE_OTEL=false
+OTLP_TRACE_ENDPOINT=
+OTLP_METRIC_ENDPOINT=
 OTLP_BASE_ENDPOINT=http://localhost:4318
 OTLP_API_KEY=
 OTEL_EXPORTER_OTLP_PROTOCOL=

api/Dockerfile

@@ -47,6 +47,8 @@ RUN \
         curl nodejs libgmp-dev libmpfr-dev libmpc-dev \
         # For Security
         expat libldap-2.5-0 perl libsqlite3-0 zlib1g \
+        # install fonts to support the use of tools like pypdfium2
+        fonts-noto-cjk \
         # install a package to improve the accuracy of guessing mime type and file extension
         media-types \
         # install libmagic to support the use of python-magic guess MIMETYPE

api/commands.py

@@ -2,19 +2,22 @@ import base64
 import json
 import logging
 import secrets
-from typing import Optional
+from typing import Any, Optional
 
 import click
 from flask import current_app
+from pydantic import TypeAdapter
 from sqlalchemy import select
 from werkzeug.exceptions import NotFound
 
 from configs import dify_config
 from constants.languages import languages
+from core.plugin.entities.plugin import ToolProviderID
 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.utils.system_oauth_encryption import encrypt_system_oauth_params
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -27,6 +30,7 @@ from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, D
 from models.dataset import Document as DatasetDocument
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation
 from models.provider import Provider, ProviderModel
+from models.tools import ToolOAuthSystemClient
 from services.account_service import AccountService, RegisterService, TenantService
 from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpiredLogs
 from services.plugin.data_migration import PluginDataMigration
@@ -1155,3 +1159,49 @@ def remove_orphaned_files_on_storage(force: bool):
         click.echo(click.style(f"Removed {removed_files} orphaned files without errors.", fg="green"))
     else:
         click.echo(click.style(f"Removed {removed_files} orphaned files, with {error_files} errors.", fg="yellow"))
+
+
+@click.command("setup-system-tool-oauth-client", help="Setup system tool oauth client.")
+@click.option("--provider", prompt=True, help="Provider name")
+@click.option("--client-params", prompt=True, help="Client Params")
+def setup_system_tool_oauth_client(provider, client_params):
+    """
+    Setup system tool oauth client
+    """
+    provider_id = ToolProviderID(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(ToolOAuthSystemClient)
+        .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 = ToolOAuthSystemClient(
+        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"))

api/configs/feature/__init__.py

@@ -31,6 +31,15 @@ class SecurityConfig(BaseSettings):
         description="Duration in minutes for which a password reset token remains valid",
         default=5,
     )
+    CHANGE_EMAIL_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
+        description="Duration in minutes for which a change email token remains valid",
+        default=5,
+    )
+
+    OWNER_TRANSFER_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
+        description="Duration in minutes for which a owner transfer token remains valid",
+        default=5,
+    )
 
     LOGIN_DISABLED: bool = Field(
         description="Whether to disable login checks",
@@ -614,6 +623,16 @@ class AuthConfig(BaseSettings):
         default=86400,
     )
 
+    CHANGE_EMAIL_LOCKOUT_DURATION: PositiveInt = Field(
+        description="Time (in seconds) a user must wait before retrying change email after exceeding the rate limit.",
+        default=86400,
+    )
+
+    OWNER_TRANSFER_LOCKOUT_DURATION: PositiveInt = Field(
+        description="Time (in seconds) a user must wait before retrying owner transfer after exceeding the rate limit.",
+        default=86400,
+    )
+
 
 class ModerationConfig(BaseSettings):
     """

api/configs/middleware/__init__.py

@@ -85,6 +85,11 @@ class VectorStoreConfig(BaseSettings):
         default=False,
     )
 
+    VECTOR_INDEX_NAME_PREFIX: Optional[str] = Field(
+        description="Prefix used to create collection name in vector database",
+        default="Vector_index",
+    )
+
 
 class KeywordStoreConfig(BaseSettings):
     KEYWORD_STORE: str = Field(
@@ -211,7 +216,7 @@ class DatabaseConfig(BaseSettings):
 class CeleryConfig(DatabaseConfig):
     CELERY_BACKEND: str = Field(
         description="Backend for Celery task results. Options: 'database', 'redis'.",
-        default="database",
+        default="redis",
     )
 
     CELERY_BROKER_URL: Optional[str] = Field(

api/configs/observability/otel/otel_config.py

@@ -12,6 +12,16 @@ class OTelConfig(BaseSettings):
         default=False,
     )
 
+    OTLP_TRACE_ENDPOINT: str = Field(
+        description="OTLP trace endpoint",
+        default="",
+    )
+
+    OTLP_METRIC_ENDPOINT: str = Field(
+        description="OTLP metric endpoint",
+        default="",
+    )
+
     OTLP_BASE_ENDPOINT: str = Field(
         description="OTLP base endpoint",
         default="http://localhost:4318",

api/constants/__init__.py

@@ -1,6 +1,7 @@
 from configs import dify_config
 
 HIDDEN_VALUE = "[__HIDDEN__]"
+UNKNOWN_VALUE = "[__UNKNOWN__]"
 UUID_NIL = "00000000-0000-0000-0000-000000000000"
 
 DEFAULT_FILE_NUMBER_LIMITS = 3

api/controllers/console/app/conversation.py

@@ -1,4 +1,4 @@
-from datetime import UTC, datetime
+from datetime import datetime
 
 import pytz  # pip install pytz
 from flask_login import current_user
@@ -19,6 +19,7 @@ from fields.conversation_fields import (
     conversation_pagination_fields,
     conversation_with_summary_pagination_fields,
 )
+from libs.datetime_utils import naive_utc_now
 from libs.helper import DatetimeString
 from libs.login import login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
@@ -315,7 +316,7 @@ def _get_conversation(app_model, conversation_id):
         raise NotFound("Conversation Not Exists.")
 
     if not conversation.read_at:
-        conversation.read_at = datetime.now(UTC).replace(tzinfo=None)
+        conversation.read_at = naive_utc_now()
         conversation.read_account_id = current_user.id
         db.session.commit()
 

api/controllers/console/app/message.py

@@ -5,6 +5,7 @@ from flask_restful import Resource, fields, marshal_with, reqparse
 from flask_restful.inputs import int_range
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
+import services
 from controllers.console import api
 from controllers.console.app.error import (
     CompletionRequestError,
@@ -27,7 +28,7 @@ from fields.conversation_fields import annotation_fields, message_detail_fields
 from libs.helper import uuid_value
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.login import login_required
-from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
+from models.model import AppMode, Conversation, Message, MessageAnnotation
 from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
@@ -124,33 +125,16 @@ class MessageFeedbackApi(Resource):
         parser.add_argument("rating", type=str, choices=["like", "dislike", None], location="json")
         args = parser.parse_args()
 
-        message_id = str(args["message_id"])
-
-        message = db.session.query(Message).filter(Message.id == message_id, Message.app_id == app_model.id).first()
-
-        if not message:
-            raise NotFound("Message Not Exists.")
-
-        feedback = message.admin_feedback
-
-        if not args["rating"] and feedback:
-            db.session.delete(feedback)
-        elif args["rating"] and feedback:
-            feedback.rating = args["rating"]
-        elif not args["rating"] and not feedback:
-            raise ValueError("rating cannot be None when feedback not exists")
-        else:
-            feedback = MessageFeedback(
-                app_id=app_model.id,
-                conversation_id=message.conversation_id,
-                message_id=message.id,
-                rating=args["rating"],
-                from_source="admin",
-                from_account_id=current_user.id,
+        try:
+            MessageService.create_feedback(
+                app_model=app_model,
+                message_id=str(args["message_id"]),
+                user=current_user,
+                rating=args.get("rating"),
+                content=None,
             )
-            db.session.add(feedback)
-
-        db.session.commit()
+        except services.errors.message.MessageNotExistsError:
+            raise NotFound("Message Not Exists.")
 
         return {"result": "success"}
 

api/controllers/console/app/site.py

@@ -1,5 +1,3 @@
-from datetime import UTC, datetime
-
 from flask_login import current_user
 from flask_restful import Resource, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden, NotFound
@@ -10,6 +8,7 @@ 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.app_fields import app_site_fields
+from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import Site
 
@@ -77,7 +76,7 @@ class AppSite(Resource):
                 setattr(site, attr_name, value)
 
         site.updated_by = current_user.id
-        site.updated_at = datetime.now(UTC).replace(tzinfo=None)
+        site.updated_at = naive_utc_now()
         db.session.commit()
 
         return site
@@ -101,7 +100,7 @@ class AppSiteAccessTokenReset(Resource):
 
         site.code = Site.generate_code(16)
         site.updated_by = current_user.id
-        site.updated_at = datetime.now(UTC).replace(tzinfo=None)
+        site.updated_at = naive_utc_now()
         db.session.commit()
 
         return site

api/controllers/console/auth/activate.py

@@ -1,5 +1,3 @@
-import datetime
-
 from flask import request
 from flask_restful import Resource, reqparse
 
@@ -7,6 +5,7 @@ from constants.languages import supported_language
 from controllers.console import api
 from controllers.console.error import AlreadyActivateError
 from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
 from libs.helper import StrLen, email, extract_remote_ip, timezone
 from models.account import AccountStatus
 from services.account_service import AccountService, RegisterService
@@ -65,7 +64,7 @@ class ActivateApi(Resource):
         account.timezone = args["timezone"]
         account.interface_theme = "light"
         account.status = AccountStatus.ACTIVE.value
-        account.initialized_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
+        account.initialized_at = naive_utc_now()
         db.session.commit()
 
         token_pair = AccountService.login(account, ip_address=extract_remote_ip(request))

api/controllers/console/auth/error.py

@@ -27,7 +27,19 @@ class InvalidTokenError(BaseHTTPException):
 
 class PasswordResetRateLimitExceededError(BaseHTTPException):
     error_code = "password_reset_rate_limit_exceeded"
-    description = "Too many password reset emails have been sent. Please try again in 1 minutes."
+    description = "Too many password reset emails have been sent. Please try again in 1 minute."
+    code = 429
+
+
+class EmailChangeRateLimitExceededError(BaseHTTPException):
+    error_code = "email_change_rate_limit_exceeded"
+    description = "Too many email change emails have been sent. Please try again in 1 minute."
+    code = 429
+
+
+class OwnerTransferRateLimitExceededError(BaseHTTPException):
+    error_code = "owner_transfer_rate_limit_exceeded"
+    description = "Too many owner transfer emails have been sent. Please try again in 1 minute."
     code = 429
 
 
@@ -65,3 +77,39 @@ class EmailPasswordResetLimitError(BaseHTTPException):
     error_code = "email_password_reset_limit"
     description = "Too many failed password reset attempts. Please try again in 24 hours."
     code = 429
+
+
+class EmailChangeLimitError(BaseHTTPException):
+    error_code = "email_change_limit"
+    description = "Too many failed email change attempts. Please try again in 24 hours."
+    code = 429
+
+
+class EmailAlreadyInUseError(BaseHTTPException):
+    error_code = "email_already_in_use"
+    description = "A user with this email already exists."
+    code = 400
+
+
+class OwnerTransferLimitError(BaseHTTPException):
+    error_code = "owner_transfer_limit"
+    description = "Too many failed owner transfer attempts. Please try again in 24 hours."
+    code = 429
+
+
+class NotOwnerError(BaseHTTPException):
+    error_code = "not_owner"
+    description = "You are not the owner of the workspace."
+    code = 400
+
+
+class CannotTransferOwnerToSelfError(BaseHTTPException):
+    error_code = "cannot_transfer_owner_to_self"
+    description = "You cannot transfer ownership to yourself."
+    code = 400
+
+
+class MemberNotInTenantError(BaseHTTPException):
+    error_code = "member_not_in_tenant"
+    description = "The member is not in the workspace."
+    code = 400

api/controllers/console/auth/oauth.py

@@ -1,5 +1,4 @@
 import logging
-from datetime import UTC, datetime
 from typing import Optional
 
 import requests
@@ -13,6 +12,7 @@ from configs import dify_config
 from constants.languages import languages
 from events.tenant_event import tenant_was_created
 from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
 from libs.helper import extract_remote_ip
 from libs.oauth import GitHubOAuth, GoogleOAuth, OAuthUserInfo
 from models import Account
@@ -110,7 +110,7 @@ class OAuthCallback(Resource):
 
         if account.status == AccountStatus.PENDING.value:
             account.status = AccountStatus.ACTIVE.value
-            account.initialized_at = datetime.now(UTC).replace(tzinfo=None)
+            account.initialized_at = naive_utc_now()
             db.session.commit()
 
         try:

api/controllers/console/datasets/data_source.py

@@ -1,4 +1,3 @@
-import datetime
 import json
 
 from flask import request
@@ -15,6 +14,7 @@ from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.notion_extractor import NotionExtractor
 from extensions.ext_database import db
 from fields.data_source_fields import integrate_list_fields, integrate_notion_info_list_fields
+from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import DataSourceOauthBinding, Document
 from services.dataset_service import DatasetService, DocumentService
@@ -88,7 +88,7 @@ class DataSourceApi(Resource):
         if action == "enable":
             if data_source_binding.disabled:
                 data_source_binding.disabled = False
-                data_source_binding.updated_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
+                data_source_binding.updated_at = naive_utc_now()
                 db.session.add(data_source_binding)
                 db.session.commit()
             else:
@@ -97,7 +97,7 @@ class DataSourceApi(Resource):
         if action == "disable":
             if not data_source_binding.disabled:
                 data_source_binding.disabled = True
-                data_source_binding.updated_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
+                data_source_binding.updated_at = naive_utc_now()
                 db.session.add(data_source_binding)
                 db.session.commit()
             else:

api/controllers/console/datasets/datasets.py

@@ -211,10 +211,6 @@ class DatasetApi(Resource):
         else:
             data["embedding_available"] = True
 
-        if data.get("permission") == "partial_members":
-            part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
-            data.update({"partial_member_list": part_users_list})
-
         return data, 200
 
     @setup_required

api/controllers/console/datasets/datasets_document.py

@@ -1,6 +1,5 @@
 import logging
 from argparse import ArgumentTypeError
-from datetime import UTC, datetime
 from typing import cast
 
 from flask import request
@@ -49,6 +48,7 @@ from fields.document_fields import (
     document_status_fields,
     document_with_segments_fields,
 )
+from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import Dataset, DatasetProcessRule, Document, DocumentSegment, UploadFile
 from services.dataset_service import DatasetService, DocumentService
@@ -750,7 +750,7 @@ class DocumentProcessingApi(DocumentResource):
                 raise InvalidActionError("Document not in indexing state.")
 
             document.paused_by = current_user.id
-            document.paused_at = datetime.now(UTC).replace(tzinfo=None)
+            document.paused_at = naive_utc_now()
             document.is_paused = True
             db.session.commit()
 
@@ -830,7 +830,7 @@ class DocumentMetadataApi(DocumentResource):
                     document.doc_metadata[key] = value
 
         document.doc_type = doc_type
-        document.updated_at = datetime.now(UTC).replace(tzinfo=None)
+        document.updated_at = naive_utc_now()
         db.session.commit()
 
         return {"result": "success", "message": "Document metadata updated."}, 200

api/controllers/console/datasets/error.py

@@ -25,12 +25,6 @@ class UnsupportedFileTypeError(BaseHTTPException):
     code = 415
 
 
-class HighQualityDatasetOnlyError(BaseHTTPException):
-    error_code = "high_quality_dataset_only"
-    description = "Current operation only supports 'high-quality' datasets."
-    code = 400
-
-
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."

api/controllers/console/datasets/website.py

@@ -4,7 +4,7 @@ from controllers.console import api
 from controllers.console.datasets.error import WebsiteCrawlError
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
-from services.website_service import WebsiteService
+from services.website_service import WebsiteCrawlApiRequest, WebsiteCrawlStatusApiRequest, WebsiteService
 
 
 class WebsiteCrawlApi(Resource):
@@ -24,10 +24,16 @@ class WebsiteCrawlApi(Resource):
         parser.add_argument("url", type=str, required=True, nullable=True, location="json")
         parser.add_argument("options", type=dict, required=True, nullable=True, location="json")
         args = parser.parse_args()
-        WebsiteService.document_create_args_validate(args)
-        # crawl url
+
+        # Create typed request and validate
         try:
-            result = WebsiteService.crawl_url(args)
+            api_request = WebsiteCrawlApiRequest.from_args(args)
+        except ValueError as e:
+            raise WebsiteCrawlError(str(e))
+
+        # Crawl URL using typed request
+        try:
+            result = WebsiteService.crawl_url(api_request)
         except Exception as e:
             raise WebsiteCrawlError(str(e))
         return result, 200
@@ -43,9 +49,16 @@ class WebsiteCrawlStatusApi(Resource):
             "provider", type=str, choices=["firecrawl", "watercrawl", "jinareader"], required=True, location="args"
         )
         args = parser.parse_args()
-        # get crawl status
+
+        # Create typed request and validate
         try:
-            result = WebsiteService.get_crawl_status(job_id, args["provider"])
+            api_request = WebsiteCrawlStatusApiRequest.from_args(args, job_id)
+        except ValueError as e:
+            raise WebsiteCrawlError(str(e))
+
+        # Get crawl status using typed request
+        try:
+            result = WebsiteService.get_crawl_status_typed(api_request)
         except Exception as e:
             raise WebsiteCrawlError(str(e))
         return result, 200

api/controllers/console/explore/completion.py

@@ -1,5 +1,4 @@
 import logging
-from datetime import UTC, datetime
 
 from flask_login import current_user
 from flask_restful import reqparse
@@ -27,6 +26,7 @@ from core.errors.error import (
 from core.model_runtime.errors.invoke import InvokeError
 from extensions.ext_database import db
 from libs import helper
+from libs.datetime_utils import naive_utc_now
 from libs.helper import uuid_value
 from models.model import AppMode
 from services.app_generate_service import AppGenerateService
@@ -51,7 +51,7 @@ class CompletionApi(InstalledAppResource):
         streaming = args["response_mode"] == "streaming"
         args["auto_generate_name"] = False
 
-        installed_app.last_used_at = datetime.now(UTC).replace(tzinfo=None)
+        installed_app.last_used_at = naive_utc_now()
         db.session.commit()
 
         try:
@@ -111,7 +111,7 @@ class ChatApi(InstalledAppResource):
 
         args["auto_generate_name"] = False
 
-        installed_app.last_used_at = datetime.now(UTC).replace(tzinfo=None)
+        installed_app.last_used_at = naive_utc_now()
         db.session.commit()
 
         try:

api/controllers/console/explore/installed_app.py

@@ -1,5 +1,4 @@
 import logging
-from datetime import UTC, datetime
 from typing import Any
 
 from flask import request
@@ -13,6 +12,7 @@ from controllers.console.explore.wraps import InstalledAppResource
 from controllers.console.wraps import account_initialization_required, cloud_edition_billing_resource_check
 from extensions.ext_database import db
 from fields.installed_app_fields import installed_app_list_fields
+from libs.datetime_utils import naive_utc_now
 from libs.login import login_required
 from models import App, InstalledApp, RecommendedApp
 from services.account_service import TenantService
@@ -122,7 +122,7 @@ class InstalledAppsListApi(Resource):
                 tenant_id=current_tenant_id,
                 app_owner_tenant_id=app.tenant_id,
                 is_pinned=False,
-                last_used_at=datetime.now(UTC).replace(tzinfo=None),
+                last_used_at=naive_utc_now(),
             )
             db.session.add(new_installed_app)
             db.session.commit()

api/controllers/console/workspace/account.py

@@ -1,13 +1,21 @@
-import datetime
-
 import pytz
 from flask import request
 from flask_login import current_user
 from flask_restful import Resource, fields, marshal_with, reqparse
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from configs import dify_config
 from constants.languages import supported_language
 from controllers.console import api
+from controllers.console.auth.error import (
+    EmailAlreadyInUseError,
+    EmailChangeLimitError,
+    EmailCodeError,
+    InvalidEmailError,
+    InvalidTokenError,
+)
+from controllers.console.error import AccountNotFound, EmailSendIpLimitError
 from controllers.console.workspace.error import (
     AccountAlreadyInitedError,
     CurrentPasswordIncorrectError,
@@ -18,15 +26,18 @@ from controllers.console.workspace.error import (
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_enabled,
+    enable_change_email,
     enterprise_license_required,
     only_edition_cloud,
     setup_required,
 )
 from extensions.ext_database import db
 from fields.member_fields import account_fields
-from libs.helper import TimestampField, timezone
+from libs.datetime_utils import naive_utc_now
+from libs.helper import TimestampField, email, extract_remote_ip, timezone
 from libs.login import login_required
 from models import AccountIntegrate, InvitationCode
+from models.account import Account
 from services.account_service import AccountService
 from services.billing_service import BillingService
 from services.errors.account import CurrentPasswordIncorrectError as ServiceCurrentPasswordIncorrectError
@@ -68,7 +79,7 @@ class AccountInitApi(Resource):
                 raise InvalidInvitationCodeError()
 
             invitation_code.status = "used"
-            invitation_code.used_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
+            invitation_code.used_at = naive_utc_now()
             invitation_code.used_by_tenant_id = account.current_tenant_id
             invitation_code.used_by_account_id = account.id
 
@@ -76,7 +87,7 @@ class AccountInitApi(Resource):
         account.timezone = args["timezone"]
         account.interface_theme = "light"
         account.status = "active"
-        account.initialized_at = datetime.datetime.now(datetime.UTC).replace(tzinfo=None)
+        account.initialized_at = naive_utc_now()
         db.session.commit()
 
         return {"result": "success"}
@@ -369,6 +380,134 @@ class EducationAutoCompleteApi(Resource):
         return BillingService.EducationIdentity.autocomplete(args["keywords"], args["page"], args["limit"])
 
 
+class ChangeEmailSendEmailApi(Resource):
+    @enable_change_email
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=email, required=True, location="json")
+        parser.add_argument("language", type=str, required=False, location="json")
+        parser.add_argument("phase", type=str, required=False, location="json")
+        parser.add_argument("token", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        ip_address = extract_remote_ip(request)
+        if AccountService.is_email_send_ip_limit(ip_address):
+            raise EmailSendIpLimitError()
+
+        if args["language"] is not None and args["language"] == "zh-Hans":
+            language = "zh-Hans"
+        else:
+            language = "en-US"
+        account = None
+        user_email = args["email"]
+        if args["phase"] is not None and args["phase"] == "new_email":
+            if args["token"] is None:
+                raise InvalidTokenError()
+
+            reset_data = AccountService.get_change_email_data(args["token"])
+            if reset_data is None:
+                raise InvalidTokenError()
+            user_email = reset_data.get("email", "")
+
+            if user_email != current_user.email:
+                raise InvalidEmailError()
+        else:
+            with Session(db.engine) as session:
+                account = session.execute(select(Account).filter_by(email=args["email"])).scalar_one_or_none()
+            if account is None:
+                raise AccountNotFound()
+
+        token = AccountService.send_change_email_email(
+            account=account, email=args["email"], old_email=user_email, language=language, phase=args["phase"]
+        )
+        return {"result": "success", "data": token}
+
+
+class ChangeEmailCheckApi(Resource):
+    @enable_change_email
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=email, required=True, location="json")
+        parser.add_argument("code", type=str, required=True, location="json")
+        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        user_email = args["email"]
+
+        is_change_email_error_rate_limit = AccountService.is_change_email_error_rate_limit(args["email"])
+        if is_change_email_error_rate_limit:
+            raise EmailChangeLimitError()
+
+        token_data = AccountService.get_change_email_data(args["token"])
+        if token_data is None:
+            raise InvalidTokenError()
+
+        if user_email != token_data.get("email"):
+            raise InvalidEmailError()
+
+        if args["code"] != token_data.get("code"):
+            AccountService.add_change_email_error_rate_limit(args["email"])
+            raise EmailCodeError()
+
+        # Verified, revoke the first token
+        AccountService.revoke_change_email_token(args["token"])
+
+        # Refresh token data by generating a new token
+        _, new_token = AccountService.generate_change_email_token(
+            user_email, code=args["code"], old_email=token_data.get("old_email"), additional_data={}
+        )
+
+        AccountService.reset_change_email_error_rate_limit(args["email"])
+        return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
+
+
+class ChangeEmailResetApi(Resource):
+    @enable_change_email
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(account_fields)
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("new_email", type=email, required=True, location="json")
+        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        reset_data = AccountService.get_change_email_data(args["token"])
+        if not reset_data:
+            raise InvalidTokenError()
+
+        AccountService.revoke_change_email_token(args["token"])
+
+        if not AccountService.check_email_unique(args["new_email"]):
+            raise EmailAlreadyInUseError()
+
+        old_email = reset_data.get("old_email", "")
+        if current_user.email != old_email:
+            raise AccountNotFound()
+
+        updated_account = AccountService.update_account(current_user, email=args["new_email"])
+
+        return updated_account
+
+
+class CheckEmailUnique(Resource):
+    @setup_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=email, required=True, location="json")
+        args = parser.parse_args()
+        if not AccountService.check_email_unique(args["email"]):
+            raise EmailAlreadyInUseError()
+        return {"result": "success"}
+
+
 # Register API resources
 api.add_resource(AccountInitApi, "/account/init")
 api.add_resource(AccountProfileApi, "/account/profile")
@@ -385,5 +524,10 @@ api.add_resource(AccountDeleteUpdateFeedbackApi, "/account/delete/feedback")
 api.add_resource(EducationVerifyApi, "/account/education/verify")
 api.add_resource(EducationApi, "/account/education")
 api.add_resource(EducationAutoCompleteApi, "/account/education/autocomplete")
+# Change email
+api.add_resource(ChangeEmailSendEmailApi, "/account/change-email")
+api.add_resource(ChangeEmailCheckApi, "/account/change-email/validity")
+api.add_resource(ChangeEmailResetApi, "/account/change-email/reset")
+api.add_resource(CheckEmailUnique, "/account/change-email/check-email-unique")
 # api.add_resource(AccountEmailApi, '/account/email')
 # api.add_resource(AccountEmailVerifyApi, '/account/email-verify')

api/controllers/console/workspace/error.py

@@ -13,12 +13,6 @@ class CurrentPasswordIncorrectError(BaseHTTPException):
     code = 400
 
 
-class ProviderRequestFailedError(BaseHTTPException):
-    error_code = "provider_request_failed"
-    description = None
-    code = 400
-
-
 class InvalidInvitationCodeError(BaseHTTPException):
     error_code = "invalid_invitation_code"
     description = "Invalid invitation code."

api/controllers/console/workspace/members.py

@@ -1,22 +1,34 @@
 from urllib import parse
 
+from flask import request
 from flask_login import current_user
 from flask_restful import Resource, abort, marshal_with, reqparse
 
 import services
 from configs import dify_config
 from controllers.console import api
-from controllers.console.error import WorkspaceMembersLimitExceeded
+from controllers.console.auth.error import (
+    CannotTransferOwnerToSelfError,
+    EmailCodeError,
+    InvalidEmailError,
+    InvalidTokenError,
+    MemberNotInTenantError,
+    NotOwnerError,
+    OwnerTransferLimitError,
+)
+from controllers.console.error import EmailSendIpLimitError, WorkspaceMembersLimitExceeded
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
+    is_allow_transfer_owner,
     setup_required,
 )
 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 models.account import Account, TenantAccountRole
-from services.account_service import RegisterService, TenantService
+from services.account_service import AccountService, RegisterService, TenantService
 from services.errors.account import AccountAlreadyInTenantError
 from services.feature_service import FeatureService
 
@@ -156,8 +168,146 @@ class DatasetOperatorMemberListApi(Resource):
         return {"result": "success", "accounts": members}, 200
 
 
+class SendOwnerTransferEmailApi(Resource):
+    """Send owner transfer email."""
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @is_allow_transfer_owner
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("language", type=str, required=False, location="json")
+        args = parser.parse_args()
+        ip_address = extract_remote_ip(request)
+        if AccountService.is_email_send_ip_limit(ip_address):
+            raise EmailSendIpLimitError()
+
+        # check if the current user is the owner of the workspace
+        if not TenantService.is_owner(current_user, current_user.current_tenant):
+            raise NotOwnerError()
+
+        if args["language"] is not None and args["language"] == "zh-Hans":
+            language = "zh-Hans"
+        else:
+            language = "en-US"
+
+        email = current_user.email
+
+        token = AccountService.send_owner_transfer_email(
+            account=current_user,
+            email=email,
+            language=language,
+            workspace_name=current_user.current_tenant.name,
+        )
+
+        return {"result": "success", "data": token}
+
+
+class OwnerTransferCheckApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @is_allow_transfer_owner
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("code", type=str, required=True, location="json")
+        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        # check if the current user is the owner of the workspace
+        if not TenantService.is_owner(current_user, current_user.current_tenant):
+            raise NotOwnerError()
+
+        user_email = current_user.email
+
+        is_owner_transfer_error_rate_limit = AccountService.is_owner_transfer_error_rate_limit(user_email)
+        if is_owner_transfer_error_rate_limit:
+            raise OwnerTransferLimitError()
+
+        token_data = AccountService.get_owner_transfer_data(args["token"])
+        if token_data is None:
+            raise InvalidTokenError()
+
+        if user_email != token_data.get("email"):
+            raise InvalidEmailError()
+
+        if args["code"] != token_data.get("code"):
+            AccountService.add_owner_transfer_error_rate_limit(user_email)
+            raise EmailCodeError()
+
+        # Verified, revoke the first token
+        AccountService.revoke_owner_transfer_token(args["token"])
+
+        # Refresh token data by generating a new token
+        _, new_token = AccountService.generate_owner_transfer_token(user_email, code=args["code"], additional_data={})
+
+        AccountService.reset_owner_transfer_error_rate_limit(user_email)
+        return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
+
+
+class OwnerTransfer(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @is_allow_transfer_owner
+    def post(self, member_id):
+        parser = reqparse.RequestParser()
+        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        # check if the current user is the owner of the workspace
+        if not TenantService.is_owner(current_user, current_user.current_tenant):
+            raise NotOwnerError()
+
+        if current_user.id == str(member_id):
+            raise CannotTransferOwnerToSelfError()
+
+        transfer_token_data = AccountService.get_owner_transfer_data(args["token"])
+        if not transfer_token_data:
+            raise InvalidTokenError()
+
+        if transfer_token_data.get("email") != current_user.email:
+            raise InvalidEmailError()
+
+        AccountService.revoke_owner_transfer_token(args["token"])
+
+        member = db.session.get(Account, str(member_id))
+        if not member:
+            abort(404)
+        else:
+            member_account = member
+        if not TenantService.is_member(member_account, current_user.current_tenant):
+            raise MemberNotInTenantError()
+
+        try:
+            assert member is not None, "Member not found"
+            TenantService.update_member_role(current_user.current_tenant, member, "owner", current_user)
+
+            AccountService.send_new_owner_transfer_notify_email(
+                account=member,
+                email=member.email,
+                workspace_name=current_user.current_tenant.name,
+            )
+
+            AccountService.send_old_owner_transfer_notify_email(
+                account=current_user,
+                email=current_user.email,
+                workspace_name=current_user.current_tenant.name,
+                new_owner_email=member.email,
+            )
+
+        except Exception as e:
+            raise ValueError(str(e))
+
+        return {"result": "success"}
+
+
 api.add_resource(MemberListApi, "/workspaces/current/members")
 api.add_resource(MemberInviteEmailApi, "/workspaces/current/members/invite-email")
 api.add_resource(MemberCancelInviteApi, "/workspaces/current/members/<uuid:member_id>")
 api.add_resource(MemberUpdateRoleApi, "/workspaces/current/members/<uuid:member_id>/update-role")
 api.add_resource(DatasetOperatorMemberListApi, "/workspaces/current/dataset-operators")
+# owner transfer
+api.add_resource(SendOwnerTransferEmailApi, "/workspaces/current/members/send-owner-transfer-confirm-email")
+api.add_resource(OwnerTransferCheckApi, "/workspaces/current/members/owner-transfer-check")
+api.add_resource(OwnerTransfer, "/workspaces/current/members/<uuid:member_id>/owner-transfer")

api/controllers/console/workspace/tool_providers.py

@@ -1,26 +1,35 @@
 import io
 from urllib.parse import urlparse
 
-from flask import redirect, send_file
+from flask import make_response, redirect, request, send_file
 from flask_login import current_user
-from flask_restful import Resource, reqparse
-from sqlalchemy.orm import Session
+from flask_restful import (
+    Resource,
+    reqparse,
+)
 from werkzeug.exceptions import Forbidden
 
 from configs import dify_config
 from controllers.console import api
-from controllers.console.wraps import account_initialization_required, enterprise_license_required, setup_required
+from controllers.console.wraps import (
+    account_initialization_required,
+    enterprise_license_required,
+    setup_required,
+)
 from core.mcp.auth.auth_flow import auth, handle_callback
 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 extensions.ext_database import db
-from libs.helper import alphanumeric, uuid_value
+from core.plugin.entities.plugin import ToolProviderID
+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 services.plugin.oauth_service import OAuthProxyService
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService
-from services.tools.mcp_tools_mange_service import MCPToolManageService
+from services.tools.mcp_tools_manage_service import MCPToolManageService
 from services.tools.tool_labels_service import ToolLabelsService
 from services.tools.tools_manage_service import ToolCommonService
 from services.tools.tools_transform_service import ToolTransformService
@@ -89,7 +98,7 @@ class ToolBuiltinProviderInfoApi(Resource):
         user_id = user.id
         tenant_id = user.current_tenant_id
 
-        return jsonable_encoder(BuiltinToolManageService.get_builtin_tool_provider_info(user_id, tenant_id, provider))
+        return jsonable_encoder(BuiltinToolManageService.get_builtin_tool_provider_info(tenant_id, provider))
 
 
 class ToolBuiltinProviderDeleteApi(Resource):
@@ -98,17 +107,47 @@ class ToolBuiltinProviderDeleteApi(Resource):
     @account_initialization_required
     def post(self, provider):
         user = current_user
-
         if not user.is_admin_or_owner:
             raise Forbidden()
 
+        tenant_id = user.current_tenant_id
+        req = reqparse.RequestParser()
+        req.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        args = req.parse_args()
+
+        return BuiltinToolManageService.delete_builtin_tool_provider(
+            tenant_id,
+            provider,
+            args["credential_id"],
+        )
+
+
+class ToolBuiltinProviderAddApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider):
+        user = current_user
+
         user_id = user.id
         tenant_id = user.current_tenant_id
 
-        return BuiltinToolManageService.delete_builtin_tool_provider(
-            user_id,
-            tenant_id,
-            provider,
+        parser = reqparse.RequestParser()
+        parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=False, location="json")
+        parser.add_argument("type", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        if args["type"] not in CredentialType.values():
+            raise ValueError(f"Invalid credential type: {args['type']}")
+
+        return BuiltinToolManageService.add_builtin_tool_provider(
+            user_id=user_id,
+            tenant_id=tenant_id,
+            provider=provider,
+            credentials=args["credentials"],
+            name=args["name"],
+            api_type=CredentialType.of(args["type"]),
         )
 
 
@@ -126,19 +165,20 @@ class ToolBuiltinProviderUpdateApi(Resource):
         tenant_id = user.current_tenant_id
 
         parser = reqparse.RequestParser()
-        parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("credential_id", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("credentials", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
 
         args = parser.parse_args()
 
-        with Session(db.engine) as session:
-            result = BuiltinToolManageService.update_builtin_tool_provider(
-                session=session,
-                user_id=user_id,
-                tenant_id=tenant_id,
-                provider_name=provider,
-                credentials=args["credentials"],
-            )
-            session.commit()
+        result = BuiltinToolManageService.update_builtin_tool_provider(
+            user_id=user_id,
+            tenant_id=tenant_id,
+            provider=provider,
+            credential_id=args["credential_id"],
+            credentials=args.get("credentials", None),
+            name=args.get("name", ""),
+        )
         return result
 
 
@@ -149,9 +189,11 @@ class ToolBuiltinProviderGetCredentialsApi(Resource):
     def get(self, provider):
         tenant_id = current_user.current_tenant_id
 
-        return BuiltinToolManageService.get_builtin_tool_provider_credentials(
-            tenant_id=tenant_id,
-            provider_name=provider,
+        return jsonable_encoder(
+            BuiltinToolManageService.get_builtin_tool_provider_credentials(
+                tenant_id=tenant_id,
+                provider_name=provider,
+            )
         )
 
 
@@ -344,12 +386,15 @@ class ToolBuiltinProviderCredentialsSchemaApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def get(self, provider):
+    def get(self, provider, credential_type):
         user = current_user
-
         tenant_id = user.current_tenant_id
 
-        return BuiltinToolManageService.list_builtin_provider_credentials_schema(provider, tenant_id)
+        return jsonable_encoder(
+            BuiltinToolManageService.list_builtin_provider_credentials_schema(
+                provider, CredentialType.of(credential_type), tenant_id
+            )
+        )
 
 
 class ToolApiProviderSchemaApi(Resource):
@@ -586,15 +631,12 @@ class ToolApiListApi(Resource):
     @account_initialization_required
     def get(self):
         user = current_user
-
-        user_id = user.id
         tenant_id = user.current_tenant_id
 
         return jsonable_encoder(
             [
                 provider.to_dict()
                 for provider in ApiToolManageService.list_api_tools(
-                    user_id,
                     tenant_id,
                 )
             ]
@@ -631,6 +673,183 @@ class ToolLabelsApi(Resource):
         return jsonable_encoder(ToolLabelsService.list_tool_labels())
 
 
+class ToolPluginOAuthApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        tool_provider = ToolProviderID(provider)
+        plugin_id = tool_provider.plugin_id
+        provider_name = tool_provider.provider_name
+
+        # todo check permission
+        user = current_user
+
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        tenant_id = user.current_tenant_id
+        oauth_client_params = BuiltinToolManageService.get_oauth_client(tenant_id=tenant_id, provider=provider)
+        if oauth_client_params is None:
+            raise Forbidden("no oauth available client config found for this tool provider")
+
+        oauth_handler = OAuthHandler()
+        context_id = OAuthProxyService.create_proxy_context(
+            user_id=current_user.id, tenant_id=tenant_id, plugin_id=plugin_id, provider=provider_name
+        )
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/tool/callback"
+        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,
+        )
+        response = make_response(jsonable_encoder(authorization_url_response))
+        response.set_cookie(
+            "context_id",
+            context_id,
+            httponly=True,
+            samesite="Lax",
+            max_age=OAuthProxyService.__MAX_AGE__,
+        )
+        return response
+
+
+class ToolOAuthCallback(Resource):
+    @setup_required
+    def get(self, provider):
+        context_id = request.cookies.get("context_id")
+        if not context_id:
+            raise Forbidden("context_id not found")
+
+        context = OAuthProxyService.use_proxy_context(context_id)
+        if context is None:
+            raise Forbidden("Invalid context_id")
+
+        tool_provider = ToolProviderID(provider)
+        plugin_id = tool_provider.plugin_id
+        provider_name = tool_provider.provider_name
+        user_id, tenant_id = context.get("user_id"), context.get("tenant_id")
+
+        oauth_handler = OAuthHandler()
+        oauth_client_params = BuiltinToolManageService.get_oauth_client(tenant_id, provider)
+        if oauth_client_params is None:
+            raise Forbidden("no oauth available client config found for this tool provider")
+
+        redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider}/tool/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("the plugin credentials failed")
+
+        # add credentials to database
+        BuiltinToolManageService.add_builtin_tool_provider(
+            user_id=user_id,
+            tenant_id=tenant_id,
+            provider=provider,
+            credentials=dict(credentials),
+            expires_at=expires_at,
+            api_type=CredentialType.OAUTH2,
+        )
+        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
+
+
+class ToolBuiltinProviderSetDefaultApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider):
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        return BuiltinToolManageService.set_default_provider(
+            tenant_id=current_user.current_tenant_id, user_id=current_user.id, provider=provider, id=args["id"]
+        )
+
+
+class ToolOAuthCustomClient(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self, provider):
+        parser = reqparse.RequestParser()
+        parser.add_argument("client_params", type=dict, required=False, nullable=True, location="json")
+        parser.add_argument("enable_oauth_custom_client", type=bool, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+
+        user = current_user
+
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        return BuiltinToolManageService.save_custom_oauth_client_params(
+            tenant_id=user.current_tenant_id,
+            provider=provider,
+            client_params=args.get("client_params", {}),
+            enable_oauth_custom_client=args.get("enable_oauth_custom_client", True),
+        )
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        return jsonable_encoder(
+            BuiltinToolManageService.get_custom_oauth_client_params(
+                tenant_id=current_user.current_tenant_id, provider=provider
+            )
+        )
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self, provider):
+        return jsonable_encoder(
+            BuiltinToolManageService.delete_custom_oauth_client_params(
+                tenant_id=current_user.current_tenant_id, provider=provider
+            )
+        )
+
+
+class ToolBuiltinProviderGetOauthClientSchemaApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        return jsonable_encoder(
+            BuiltinToolManageService.get_builtin_tool_provider_oauth_client_schema(
+                tenant_id=current_user.current_tenant_id, provider_name=provider
+            )
+        )
+
+
+class ToolBuiltinProviderGetCredentialInfoApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider):
+        tenant_id = current_user.current_tenant_id
+
+        return jsonable_encoder(
+            BuiltinToolManageService.get_builtin_tool_provider_credential_info(
+                tenant_id=tenant_id,
+                provider=provider,
+            )
+        )
+
+
 class ToolProviderMCPApi(Resource):
     @setup_required
     @login_required
@@ -794,17 +1013,33 @@ class ToolMCPCallbackApi(Resource):
 # tool provider
 api.add_resource(ToolProviderListApi, "/workspaces/current/tool-providers")
 
+# tool oauth
+api.add_resource(ToolPluginOAuthApi, "/oauth/plugin/<path:provider>/tool/authorization-url")
+api.add_resource(ToolOAuthCallback, "/oauth/plugin/<path:provider>/tool/callback")
+api.add_resource(ToolOAuthCustomClient, "/workspaces/current/tool-provider/builtin/<path:provider>/oauth/custom-client")
+
 # builtin tool provider
 api.add_resource(ToolBuiltinProviderListToolsApi, "/workspaces/current/tool-provider/builtin/<path:provider>/tools")
 api.add_resource(ToolBuiltinProviderInfoApi, "/workspaces/current/tool-provider/builtin/<path:provider>/info")
+api.add_resource(ToolBuiltinProviderAddApi, "/workspaces/current/tool-provider/builtin/<path:provider>/add")
 api.add_resource(ToolBuiltinProviderDeleteApi, "/workspaces/current/tool-provider/builtin/<path:provider>/delete")
 api.add_resource(ToolBuiltinProviderUpdateApi, "/workspaces/current/tool-provider/builtin/<path:provider>/update")
+api.add_resource(
+    ToolBuiltinProviderSetDefaultApi, "/workspaces/current/tool-provider/builtin/<path:provider>/default-credential"
+)
+api.add_resource(
+    ToolBuiltinProviderGetCredentialInfoApi, "/workspaces/current/tool-provider/builtin/<path:provider>/credential/info"
+)
 api.add_resource(
     ToolBuiltinProviderGetCredentialsApi, "/workspaces/current/tool-provider/builtin/<path:provider>/credentials"
 )
 api.add_resource(
     ToolBuiltinProviderCredentialsSchemaApi,
-    "/workspaces/current/tool-provider/builtin/<path:provider>/credentials_schema",
+    "/workspaces/current/tool-provider/builtin/<path:provider>/credential/schema/<path:credential_type>",
+)
+api.add_resource(
+    ToolBuiltinProviderGetOauthClientSchemaApi,
+    "/workspaces/current/tool-provider/builtin/<path:provider>/oauth/client-schema",
 )
 api.add_resource(ToolBuiltinProviderIconApi, "/workspaces/current/tool-provider/builtin/<path:provider>/icon")
 

api/controllers/console/wraps.py

@@ -235,3 +235,29 @@ def email_password_login_enabled(view):
         abort(403)
 
     return decorated
+
+
+def enable_change_email(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_system_features()
+        if features.enable_change_email:
+            return view(*args, **kwargs)
+
+        # otherwise, return 403
+        abort(403)
+
+    return decorated
+
+
+def is_allow_transfer_owner(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_features(current_user.current_tenant_id)
+        if features.is_allow_transfer_workspace:
+            return view(*args, **kwargs)
+
+        # otherwise, return 403
+        abort(403)
+
+    return decorated

api/controllers/inner_api/plugin/plugin.py

@@ -175,6 +175,7 @@ class PluginInvokeToolApi(Resource):
                     provider=payload.provider,
                     tool_name=payload.tool,
                     tool_parameters=payload.tool_parameters,
+                    credential_id=payload.credential_id,
                 ),
             )
 

api/controllers/service_api/dataset/error.py

@@ -25,12 +25,6 @@ class UnsupportedFileTypeError(BaseHTTPException):
     code = 415
 
 
-class HighQualityDatasetOnlyError(BaseHTTPException):
-    error_code = "high_quality_dataset_only"
-    description = "Current operation only supports 'high-quality' datasets."
-    code = 400
-
-
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."

api/controllers/service_api/wraps.py

@@ -1,6 +1,6 @@
 import time
 from collections.abc import Callable
-from datetime import UTC, datetime, timedelta
+from datetime import timedelta
 from enum import Enum
 from functools import wraps
 from typing import Optional
@@ -15,6 +15,7 @@ from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
+from libs.datetime_utils import naive_utc_now
 from libs.login import _get_user
 from models.account import Account, Tenant, TenantAccountJoin, TenantStatus
 from models.dataset import Dataset, RateLimitLog
@@ -256,7 +257,7 @@ def validate_and_get_api_token(scope: str | None = None):
     if auth_scheme != "bearer":
         raise Unauthorized("Authorization scheme must be 'Bearer'")
 
-    current_time = datetime.now(UTC).replace(tzinfo=None)
+    current_time = naive_utc_now()
     cutoff_time = current_time - timedelta(minutes=1)
     with Session(db.engine, expire_on_commit=False) as session:
         update_stmt = (

api/core/agent/entities.py

@@ -16,6 +16,7 @@ class AgentToolEntity(BaseModel):
     tool_name: str
     tool_parameters: dict[str, Any] = Field(default_factory=dict)
     plugin_unique_identifier: str | None = None
+    credential_id: str | None = None
 
 
 class AgentPromptEntity(BaseModel):

api/core/agent/plugin_entities.py

@@ -41,6 +41,7 @@ class AgentStrategyParameter(PluginParameter):
         APP_SELECTOR = CommonParameterType.APP_SELECTOR.value
         MODEL_SELECTOR = CommonParameterType.MODEL_SELECTOR.value
         TOOLS_SELECTOR = CommonParameterType.TOOLS_SELECTOR.value
+        ANY = CommonParameterType.ANY.value
 
         # deprecated, should not use.
         SYSTEM_FILES = CommonParameterType.SYSTEM_FILES.value

api/core/agent/strategy/base.py

@@ -4,6 +4,7 @@ from typing import Any, Optional
 
 from core.agent.entities import AgentInvokeMessage
 from core.agent.plugin_entities import AgentStrategyParameter
+from core.plugin.entities.request import InvokeCredentials
 
 
 class BaseAgentStrategy(ABC):
@@ -18,11 +19,12 @@ class BaseAgentStrategy(ABC):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
+        credentials: Optional[InvokeCredentials] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         """
         Invoke the agent strategy.
         """
-        yield from self._invoke(params, user_id, conversation_id, app_id, message_id)
+        yield from self._invoke(params, user_id, conversation_id, app_id, message_id, credentials)
 
     def get_parameters(self) -> Sequence[AgentStrategyParameter]:
         """
@@ -38,5 +40,6 @@ class BaseAgentStrategy(ABC):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
+        credentials: Optional[InvokeCredentials] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         pass

api/core/agent/strategy/plugin.py

@@ -4,6 +4,7 @@ from typing import Any, Optional
 from core.agent.entities import AgentInvokeMessage
 from core.agent.plugin_entities import AgentStrategyEntity, AgentStrategyParameter
 from core.agent.strategy.base import BaseAgentStrategy
+from core.plugin.entities.request import InvokeCredentials, PluginInvokeContext
 from core.plugin.impl.agent import PluginAgentClient
 from core.plugin.utils.converter import convert_parameters_to_plugin_format
 
@@ -40,6 +41,7 @@ class PluginAgentStrategy(BaseAgentStrategy):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
+        credentials: Optional[InvokeCredentials] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         """
         Invoke the agent strategy.
@@ -58,4 +60,5 @@ class PluginAgentStrategy(BaseAgentStrategy):
             conversation_id=conversation_id,
             app_id=app_id,
             message_id=message_id,
+            context=PluginInvokeContext(credentials=credentials or InvokeCredentials()),
         )

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

@@ -39,6 +39,7 @@ class AgentConfigManager:
                         "provider_id": tool["provider_id"],
                         "tool_name": tool["tool_name"],
                         "tool_parameters": tool.get("tool_parameters", {}),
+                        "credential_id": tool.get("credential_id", None),
                     }
 
                     agent_tools.append(AgentToolEntity(**agent_tool_properties))

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

@@ -17,7 +17,8 @@ from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.advanced_chat.app_runner import AdvancedChatAppRunner
 from core.app.apps.advanced_chat.generate_response_converter import AdvancedChatAppGenerateResponseConverter
 from core.app.apps.advanced_chat.generate_task_pipeline import AdvancedChatAppGenerateTaskPipeline
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, InvokeFrom

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

@@ -15,7 +15,8 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.agent_chat.app_config_manager import AgentChatAppConfigManager
 from core.app.apps.agent_chat.app_runner import AgentChatAppRunner
 from core.app.apps.agent_chat.generate_response_converter import AgentChatAppGenerateResponseConverter
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, InvokeFrom

api/core/app/apps/base_app_queue_manager.py

@@ -169,7 +169,3 @@ class AppQueueManager:
                 raise TypeError(
                     "Critical Error: Passing SQLAlchemy Model instances that cause thread safety issues is not allowed."
                 )
-
-
-class GenerateTaskStoppedError(Exception):
-    pass

api/core/app/apps/base_app_runner.py

@@ -38,69 +38,6 @@ _logger = logging.getLogger(__name__)
 
 
 class AppRunner:
-    def get_pre_calculate_rest_tokens(
-        self,
-        app_record: App,
-        model_config: ModelConfigWithCredentialsEntity,
-        prompt_template_entity: PromptTemplateEntity,
-        inputs: Mapping[str, str],
-        files: Sequence["File"],
-        query: Optional[str] = None,
-    ) -> int:
-        """
-        Get pre calculate rest tokens
-        :param app_record: app record
-        :param model_config: model config entity
-        :param prompt_template_entity: prompt template entity
-        :param inputs: inputs
-        :param files: files
-        :param query: query
-        :return:
-        """
-        # Invoke model
-        model_instance = ModelInstance(
-            provider_model_bundle=model_config.provider_model_bundle, model=model_config.model
-        )
-
-        model_context_tokens = model_config.model_schema.model_properties.get(ModelPropertyKey.CONTEXT_SIZE)
-
-        max_tokens = 0
-        for parameter_rule in model_config.model_schema.parameter_rules:
-            if parameter_rule.name == "max_tokens" or (
-                parameter_rule.use_template and parameter_rule.use_template == "max_tokens"
-            ):
-                max_tokens = (
-                    model_config.parameters.get(parameter_rule.name)
-                    or model_config.parameters.get(parameter_rule.use_template or "")
-                ) or 0
-
-        if model_context_tokens is None:
-            return -1
-
-        if max_tokens is None:
-            max_tokens = 0
-
-        # get prompt messages without memory and context
-        prompt_messages, stop = self.organize_prompt_messages(
-            app_record=app_record,
-            model_config=model_config,
-            prompt_template_entity=prompt_template_entity,
-            inputs=inputs,
-            files=files,
-            query=query,
-        )
-
-        prompt_tokens = model_instance.get_llm_num_tokens(prompt_messages)
-
-        rest_tokens: int = model_context_tokens - max_tokens - prompt_tokens
-        if rest_tokens < 0:
-            raise InvokeBadRequestError(
-                "Query or prefix prompt is too long, you can reduce the prefix prompt, "
-                "or shrink the max token, or switch to a llm with a larger token limit size."
-            )
-
-        return rest_tokens
-
     def recalc_llm_max_tokens(
         self, model_config: ModelConfigWithCredentialsEntity, prompt_messages: list[PromptMessage]
     ):
@@ -181,7 +118,7 @@ class AppRunner:
         else:
             memory_config = MemoryConfig(window=MemoryConfig.WindowConfig(enabled=False))
 
-            model_mode = ModelMode.value_of(model_config.mode)
+            model_mode = ModelMode(model_config.mode)
             prompt_template: Union[CompletionModelPromptTemplate, list[ChatModelMessage]]
             if model_mode == ModelMode.COMPLETION:
                 advanced_completion_prompt_template = prompt_template_entity.advanced_completion_prompt_template

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

@@ -11,10 +11,11 @@ from configs import dify_config
 from constants import UUID_NIL
 from core.app.app_config.easy_ui_based_app.model_config.converter import ModelConfigConverter
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.chat.app_config_manager import ChatAppConfigManager
 from core.app.apps.chat.app_runner import ChatAppRunner
 from core.app.apps.chat.generate_response_converter import ChatAppGenerateResponseConverter
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import ChatAppGenerateEntity, InvokeFrom

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

@@ -10,10 +10,11 @@ from pydantic import ValidationError
 from configs import dify_config
 from core.app.app_config.easy_ui_based_app.model_config.converter import ModelConfigConverter
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.completion.app_config_manager import CompletionAppConfigManager
 from core.app.apps.completion.app_runner import CompletionAppRunner
 from core.app.apps.completion.generate_response_converter import CompletionAppGenerateResponseConverter
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import CompletionAppGenerateEntity, InvokeFrom

api/core/app/apps/exc.py

@@ -0,0 +1,2 @@
+class GenerateTaskStoppedError(Exception):
+    pass

api/core/app/apps/message_based_app_generator.py

@@ -1,12 +1,12 @@
 import json
 import logging
 from collections.abc import Generator
-from datetime import UTC, datetime
 from typing import Optional, Union, cast
 
 from core.app.app_config.entities import EasyUIBasedAppConfig, EasyUIBasedAppModelConfigFrom
 from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import (
     AdvancedChatAppGenerateEntity,
     AgentChatAppGenerateEntity,
@@ -24,6 +24,7 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBasedGenerateTaskPipeline
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
 from models import Account
 from models.enums import CreatorUserRole
 from models.model import App, AppMode, AppModelConfig, Conversation, EndUser, Message, MessageFile
@@ -183,7 +184,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
             db.session.commit()
             db.session.refresh(conversation)
         else:
-            conversation.updated_at = datetime.now(UTC).replace(tzinfo=None)
+            conversation.updated_at = naive_utc_now()
             db.session.commit()
 
         message = Message(

api/core/app/apps/message_based_app_queue_manager.py

@@ -1,4 +1,5 @@
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,

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

@@ -13,7 +13,8 @@ import contexts
 from configs import dify_config
 from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_generator import BaseAppGenerator
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager
 from core.app.apps.workflow.app_runner import WorkflowAppRunner

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

@@ -1,4 +1,5 @@
-from core.app.apps.base_app_queue_manager import AppQueueManager, GenerateTaskStoppedError, PublishFrom
+from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import (
     AppQueueEvent,

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

@@ -1,7 +1,8 @@
 import logging
 import time
-from collections.abc import Generator
-from typing import Optional, Union
+from collections.abc import Callable, Generator
+from contextlib import contextmanager
+from typing import Any, Optional, Union
 
 from sqlalchemy.orm import Session
 
@@ -13,6 +14,7 @@ from core.app.entities.app_invoke_entities import (
     WorkflowAppGenerateEntity,
 )
 from core.app.entities.queue_entities import (
+    MessageQueueMessage,
     QueueAgentLogEvent,
     QueueErrorEvent,
     QueueIterationCompletedEvent,
@@ -38,11 +40,13 @@ from core.app.entities.queue_entities import (
     QueueWorkflowPartialSuccessEvent,
     QueueWorkflowStartedEvent,
     QueueWorkflowSucceededEvent,
+    WorkflowQueueMessage,
 )
 from core.app.entities.task_entities import (
     ErrorStreamResponse,
     MessageAudioEndStreamResponse,
     MessageAudioStreamResponse,
+    PingStreamResponse,
     StreamResponse,
     TextChunkStreamResponse,
     WorkflowAppBlockingResponse,
@@ -54,6 +58,7 @@ from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTas
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.entities.workflow_execution import WorkflowExecution, WorkflowExecutionStatus, WorkflowType
+from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
@@ -246,315 +251,492 @@ class WorkflowAppGenerateTaskPipeline:
         if tts_publisher:
             yield MessageAudioEndStreamResponse(audio="", task_id=task_id)
 
+    @contextmanager
+    def _database_session(self):
+        """Context manager for database sessions."""
+        with Session(db.engine, expire_on_commit=False) as session:
+            try:
+                yield session
+                session.commit()
+            except Exception:
+                session.rollback()
+                raise
+
+    def _ensure_workflow_initialized(self) -> None:
+        """Fluent validation for workflow state."""
+        if not self._workflow_run_id:
+            raise ValueError("workflow run not initialized.")
+
+    def _ensure_graph_runtime_initialized(self, graph_runtime_state: Optional[GraphRuntimeState]) -> GraphRuntimeState:
+        """Fluent validation for graph runtime state."""
+        if not graph_runtime_state:
+            raise ValueError("graph runtime state not initialized.")
+        return graph_runtime_state
+
+    def _handle_ping_event(self, event: QueuePingEvent, **kwargs) -> Generator[PingStreamResponse, None, None]:
+        """Handle ping events."""
+        yield self._base_task_pipeline._ping_stream_response()
+
+    def _handle_error_event(self, event: QueueErrorEvent, **kwargs) -> Generator[ErrorStreamResponse, None, None]:
+        """Handle error events."""
+        err = self._base_task_pipeline._handle_error(event=event)
+        yield self._base_task_pipeline._error_to_stream_response(err)
+
+    def _handle_workflow_started_event(
+        self, event: QueueWorkflowStartedEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle workflow started events."""
+        # init workflow run
+        workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
+        self._workflow_run_id = workflow_execution.id_
+        start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution=workflow_execution,
+        )
+        yield start_resp
+
+    def _handle_node_retry_event(self, event: QueueNodeRetryEvent, **kwargs) -> Generator[StreamResponse, None, None]:
+        """Handle node retry events."""
+        self._ensure_workflow_initialized()
+
+        with self._database_session() as session:
+            workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
+                workflow_execution_id=self._workflow_run_id,
+                event=event,
+            )
+            response = self._workflow_response_converter.workflow_node_retry_to_stream_response(
+                event=event,
+                task_id=self._application_generate_entity.task_id,
+                workflow_node_execution=workflow_node_execution,
+            )
+
+        if response:
+            yield response
+
+    def _handle_node_started_event(
+        self, event: QueueNodeStartedEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle node started events."""
+        self._ensure_workflow_initialized()
+
+        workflow_node_execution = self._workflow_cycle_manager.handle_node_execution_start(
+            workflow_execution_id=self._workflow_run_id, event=event
+        )
+        node_start_response = self._workflow_response_converter.workflow_node_start_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            workflow_node_execution=workflow_node_execution,
+        )
+
+        if node_start_response:
+            yield node_start_response
+
+    def _handle_node_succeeded_event(
+        self, event: QueueNodeSucceededEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle node succeeded events."""
+        workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_success(event=event)
+        node_success_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            workflow_node_execution=workflow_node_execution,
+        )
+
+        self._save_output_for_event(event, workflow_node_execution.id)
+
+        if node_success_response:
+            yield node_success_response
+
+    def _handle_node_failed_events(
+        self,
+        event: Union[
+            QueueNodeFailedEvent, QueueNodeInIterationFailedEvent, QueueNodeInLoopFailedEvent, QueueNodeExceptionEvent
+        ],
+        **kwargs,
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle various node failure events."""
+        workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_failed(
+            event=event,
+        )
+        node_failed_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            workflow_node_execution=workflow_node_execution,
+        )
+
+        if isinstance(event, QueueNodeExceptionEvent):
+            self._save_output_for_event(event, workflow_node_execution.id)
+
+        if node_failed_response:
+            yield node_failed_response
+
+    def _handle_parallel_branch_started_event(
+        self, event: QueueParallelBranchRunStartedEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle parallel branch started events."""
+        self._ensure_workflow_initialized()
+
+        parallel_start_resp = self._workflow_response_converter.workflow_parallel_branch_start_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield parallel_start_resp
+
+    def _handle_parallel_branch_finished_events(
+        self, event: Union[QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent], **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle parallel branch finished events."""
+        self._ensure_workflow_initialized()
+
+        parallel_finish_resp = self._workflow_response_converter.workflow_parallel_branch_finished_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield parallel_finish_resp
+
+    def _handle_iteration_start_event(
+        self, event: QueueIterationStartEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle iteration start events."""
+        self._ensure_workflow_initialized()
+
+        iter_start_resp = self._workflow_response_converter.workflow_iteration_start_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield iter_start_resp
+
+    def _handle_iteration_next_event(
+        self, event: QueueIterationNextEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle iteration next events."""
+        self._ensure_workflow_initialized()
+
+        iter_next_resp = self._workflow_response_converter.workflow_iteration_next_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield iter_next_resp
+
+    def _handle_iteration_completed_event(
+        self, event: QueueIterationCompletedEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle iteration completed events."""
+        self._ensure_workflow_initialized()
+
+        iter_finish_resp = self._workflow_response_converter.workflow_iteration_completed_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield iter_finish_resp
+
+    def _handle_loop_start_event(self, event: QueueLoopStartEvent, **kwargs) -> Generator[StreamResponse, None, None]:
+        """Handle loop start events."""
+        self._ensure_workflow_initialized()
+
+        loop_start_resp = self._workflow_response_converter.workflow_loop_start_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield loop_start_resp
+
+    def _handle_loop_next_event(self, event: QueueLoopNextEvent, **kwargs) -> Generator[StreamResponse, None, None]:
+        """Handle loop next events."""
+        self._ensure_workflow_initialized()
+
+        loop_next_resp = self._workflow_response_converter.workflow_loop_next_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield loop_next_resp
+
+    def _handle_loop_completed_event(
+        self, event: QueueLoopCompletedEvent, **kwargs
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle loop completed events."""
+        self._ensure_workflow_initialized()
+
+        loop_finish_resp = self._workflow_response_converter.workflow_loop_completed_to_stream_response(
+            task_id=self._application_generate_entity.task_id,
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        yield loop_finish_resp
+
+    def _handle_workflow_succeeded_event(
+        self,
+        event: QueueWorkflowSucceededEvent,
+        *,
+        graph_runtime_state: Optional[GraphRuntimeState] = None,
+        trace_manager: Optional[TraceQueueManager] = None,
+        **kwargs,
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle workflow succeeded events."""
+        self._ensure_workflow_initialized()
+        validated_state = self._ensure_graph_runtime_initialized(graph_runtime_state)
+
+        with self._database_session() as session:
+            workflow_execution = self._workflow_cycle_manager.handle_workflow_run_success(
+                workflow_run_id=self._workflow_run_id,
+                total_tokens=validated_state.total_tokens,
+                total_steps=validated_state.node_run_steps,
+                outputs=event.outputs,
+                conversation_id=None,
+                trace_manager=trace_manager,
+            )
+
+            # save workflow app log
+            self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
+
+            workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                session=session,
+                task_id=self._application_generate_entity.task_id,
+                workflow_execution=workflow_execution,
+            )
+
+        yield workflow_finish_resp
+
+    def _handle_workflow_partial_success_event(
+        self,
+        event: QueueWorkflowPartialSuccessEvent,
+        *,
+        graph_runtime_state: Optional[GraphRuntimeState] = None,
+        trace_manager: Optional[TraceQueueManager] = None,
+        **kwargs,
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle workflow partial success events."""
+        self._ensure_workflow_initialized()
+        validated_state = self._ensure_graph_runtime_initialized(graph_runtime_state)
+
+        with self._database_session() as session:
+            workflow_execution = self._workflow_cycle_manager.handle_workflow_run_partial_success(
+                workflow_run_id=self._workflow_run_id,
+                total_tokens=validated_state.total_tokens,
+                total_steps=validated_state.node_run_steps,
+                outputs=event.outputs,
+                exceptions_count=event.exceptions_count,
+                conversation_id=None,
+                trace_manager=trace_manager,
+            )
+
+            # save workflow app log
+            self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
+
+            workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                session=session,
+                task_id=self._application_generate_entity.task_id,
+                workflow_execution=workflow_execution,
+            )
+
+        yield workflow_finish_resp
+
+    def _handle_workflow_failed_and_stop_events(
+        self,
+        event: Union[QueueWorkflowFailedEvent, QueueStopEvent],
+        *,
+        graph_runtime_state: Optional[GraphRuntimeState] = None,
+        trace_manager: Optional[TraceQueueManager] = None,
+        **kwargs,
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle workflow failed and stop events."""
+        self._ensure_workflow_initialized()
+        validated_state = self._ensure_graph_runtime_initialized(graph_runtime_state)
+
+        with self._database_session() as session:
+            workflow_execution = self._workflow_cycle_manager.handle_workflow_run_failed(
+                workflow_run_id=self._workflow_run_id,
+                total_tokens=validated_state.total_tokens,
+                total_steps=validated_state.node_run_steps,
+                status=WorkflowExecutionStatus.FAILED
+                if isinstance(event, QueueWorkflowFailedEvent)
+                else WorkflowExecutionStatus.STOPPED,
+                error_message=event.error if isinstance(event, QueueWorkflowFailedEvent) else event.get_stop_reason(),
+                conversation_id=None,
+                trace_manager=trace_manager,
+                exceptions_count=event.exceptions_count if isinstance(event, QueueWorkflowFailedEvent) else 0,
+            )
+
+            # save workflow app log
+            self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
+
+            workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                session=session,
+                task_id=self._application_generate_entity.task_id,
+                workflow_execution=workflow_execution,
+            )
+
+        yield workflow_finish_resp
+
+    def _handle_text_chunk_event(
+        self,
+        event: QueueTextChunkEvent,
+        *,
+        tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
+        queue_message: Optional[Union[WorkflowQueueMessage, MessageQueueMessage]] = None,
+        **kwargs,
+    ) -> Generator[StreamResponse, None, None]:
+        """Handle text chunk events."""
+        delta_text = event.text
+        if delta_text is None:
+            return
+
+        # only publish tts message at text chunk streaming
+        if tts_publisher and queue_message:
+            tts_publisher.publish(queue_message)
+
+        yield self._text_chunk_to_stream_response(delta_text, from_variable_selector=event.from_variable_selector)
+
+    def _handle_agent_log_event(self, event: QueueAgentLogEvent, **kwargs) -> Generator[StreamResponse, None, None]:
+        """Handle agent log events."""
+        yield self._workflow_response_converter.handle_agent_log(
+            task_id=self._application_generate_entity.task_id, event=event
+        )
+
+    def _get_event_handlers(self) -> dict[type, Callable]:
+        """Get mapping of event types to their handlers using fluent pattern."""
+        return {
+            # Basic events
+            QueuePingEvent: self._handle_ping_event,
+            QueueErrorEvent: self._handle_error_event,
+            QueueTextChunkEvent: self._handle_text_chunk_event,
+            # Workflow events
+            QueueWorkflowStartedEvent: self._handle_workflow_started_event,
+            QueueWorkflowSucceededEvent: self._handle_workflow_succeeded_event,
+            QueueWorkflowPartialSuccessEvent: self._handle_workflow_partial_success_event,
+            # Node events
+            QueueNodeRetryEvent: self._handle_node_retry_event,
+            QueueNodeStartedEvent: self._handle_node_started_event,
+            QueueNodeSucceededEvent: self._handle_node_succeeded_event,
+            # Parallel branch events
+            QueueParallelBranchRunStartedEvent: self._handle_parallel_branch_started_event,
+            # Iteration events
+            QueueIterationStartEvent: self._handle_iteration_start_event,
+            QueueIterationNextEvent: self._handle_iteration_next_event,
+            QueueIterationCompletedEvent: self._handle_iteration_completed_event,
+            # Loop events
+            QueueLoopStartEvent: self._handle_loop_start_event,
+            QueueLoopNextEvent: self._handle_loop_next_event,
+            QueueLoopCompletedEvent: self._handle_loop_completed_event,
+            # Agent events
+            QueueAgentLogEvent: self._handle_agent_log_event,
+        }
+
+    def _dispatch_event(
+        self,
+        event: Any,
+        *,
+        graph_runtime_state: Optional[GraphRuntimeState] = None,
+        tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
+        trace_manager: Optional[TraceQueueManager] = None,
+        queue_message: Optional[Union[WorkflowQueueMessage, MessageQueueMessage]] = None,
+    ) -> Generator[StreamResponse, None, None]:
+        """Dispatch events using elegant pattern matching."""
+        handlers = self._get_event_handlers()
+        event_type = type(event)
+
+        # Direct handler lookup
+        if handler := handlers.get(event_type):
+            yield from handler(
+                event,
+                graph_runtime_state=graph_runtime_state,
+                tts_publisher=tts_publisher,
+                trace_manager=trace_manager,
+                queue_message=queue_message,
+            )
+            return
+
+        # Handle node failure events with isinstance check
+        if isinstance(
+            event,
+            (
+                QueueNodeFailedEvent,
+                QueueNodeInIterationFailedEvent,
+                QueueNodeInLoopFailedEvent,
+                QueueNodeExceptionEvent,
+            ),
+        ):
+            yield from self._handle_node_failed_events(
+                event,
+                graph_runtime_state=graph_runtime_state,
+                tts_publisher=tts_publisher,
+                trace_manager=trace_manager,
+                queue_message=queue_message,
+            )
+            return
+
+        # Handle parallel branch finished events with isinstance check
+        if isinstance(event, (QueueParallelBranchRunSucceededEvent, QueueParallelBranchRunFailedEvent)):
+            yield from self._handle_parallel_branch_finished_events(
+                event,
+                graph_runtime_state=graph_runtime_state,
+                tts_publisher=tts_publisher,
+                trace_manager=trace_manager,
+                queue_message=queue_message,
+            )
+            return
+
+        # Handle workflow failed and stop events with isinstance check
+        if isinstance(event, (QueueWorkflowFailedEvent, QueueStopEvent)):
+            yield from self._handle_workflow_failed_and_stop_events(
+                event,
+                graph_runtime_state=graph_runtime_state,
+                tts_publisher=tts_publisher,
+                trace_manager=trace_manager,
+                queue_message=queue_message,
+            )
+            return
+
+        # For unhandled events, we continue (original behavior)
+        return
+
     def _process_stream_response(
         self,
         tts_publisher: Optional[AppGeneratorTTSPublisher] = None,
         trace_manager: Optional[TraceQueueManager] = None,
     ) -> Generator[StreamResponse, None, None]:
         """
-        Process stream response.
-        :return:
+        Process stream response using elegant Fluent Python patterns.
+        Maintains exact same functionality as original 44-if-statement version.
         """
+        # Initialize graph runtime state
         graph_runtime_state = None
 
         for queue_message in self._base_task_pipeline._queue_manager.listen():
             event = queue_message.event
 
-            if isinstance(event, QueuePingEvent):
-                yield self._base_task_pipeline._ping_stream_response()
-            elif isinstance(event, QueueErrorEvent):
-                err = self._base_task_pipeline._handle_error(event=event)
-                yield self._base_task_pipeline._error_to_stream_response(err)
-                break
-            elif isinstance(event, QueueWorkflowStartedEvent):
-                # override graph runtime state
-                graph_runtime_state = event.graph_runtime_state
+            match event:
+                case QueueWorkflowStartedEvent():
+                    graph_runtime_state = event.graph_runtime_state
+                    yield from self._handle_workflow_started_event(event)
 
-                # init workflow run
-                workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
-                self._workflow_run_id = workflow_execution.id_
-                start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_execution=workflow_execution,
-                )
-
-                yield start_resp
-            elif isinstance(
-                event,
-                QueueNodeRetryEvent,
-            ):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
-                        workflow_execution_id=self._workflow_run_id,
-                        event=event,
-                    )
-                    response = self._workflow_response_converter.workflow_node_retry_to_stream_response(
-                        event=event,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_node_execution=workflow_node_execution,
-                    )
-                    session.commit()
-
-                if response:
-                    yield response
-            elif isinstance(event, QueueNodeStartedEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                workflow_node_execution = self._workflow_cycle_manager.handle_node_execution_start(
-                    workflow_execution_id=self._workflow_run_id, event=event
-                )
-                node_start_response = self._workflow_response_converter.workflow_node_start_to_stream_response(
-                    event=event,
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_node_execution=workflow_node_execution,
-                )
-
-                if node_start_response:
-                    yield node_start_response
-            elif isinstance(event, QueueNodeSucceededEvent):
-                workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_success(
-                    event=event
-                )
-                node_success_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
-                    event=event,
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_node_execution=workflow_node_execution,
-                )
-
-                self._save_output_for_event(event, workflow_node_execution.id)
-
-                if node_success_response:
-                    yield node_success_response
-            elif isinstance(
-                event,
-                QueueNodeFailedEvent
-                | QueueNodeInIterationFailedEvent
-                | QueueNodeInLoopFailedEvent
-                | QueueNodeExceptionEvent,
-            ):
-                workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_failed(
-                    event=event,
-                )
-                node_failed_response = self._workflow_response_converter.workflow_node_finish_to_stream_response(
-                    event=event,
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_node_execution=workflow_node_execution,
-                )
-                if isinstance(event, QueueNodeExceptionEvent):
-                    self._save_output_for_event(event, workflow_node_execution.id)
-
-                if node_failed_response:
-                    yield node_failed_response
-
-            elif isinstance(event, QueueParallelBranchRunStartedEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                parallel_start_resp = (
-                    self._workflow_response_converter.workflow_parallel_branch_start_to_stream_response(
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_execution_id=self._workflow_run_id,
-                        event=event,
-                    )
-                )
-
-                yield parallel_start_resp
-
-            elif isinstance(event, QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                parallel_finish_resp = (
-                    self._workflow_response_converter.workflow_parallel_branch_finished_to_stream_response(
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_execution_id=self._workflow_run_id,
-                        event=event,
-                    )
-                )
-
-                yield parallel_finish_resp
-
-            elif isinstance(event, QueueIterationStartEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                iter_start_resp = self._workflow_response_converter.workflow_iteration_start_to_stream_response(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_execution_id=self._workflow_run_id,
-                    event=event,
-                )
-
-                yield iter_start_resp
-
-            elif isinstance(event, QueueIterationNextEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                iter_next_resp = self._workflow_response_converter.workflow_iteration_next_to_stream_response(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_execution_id=self._workflow_run_id,
-                    event=event,
-                )
-
-                yield iter_next_resp
-
-            elif isinstance(event, QueueIterationCompletedEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                iter_finish_resp = self._workflow_response_converter.workflow_iteration_completed_to_stream_response(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_execution_id=self._workflow_run_id,
-                    event=event,
-                )
-
-                yield iter_finish_resp
-
-            elif isinstance(event, QueueLoopStartEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                loop_start_resp = self._workflow_response_converter.workflow_loop_start_to_stream_response(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_execution_id=self._workflow_run_id,
-                    event=event,
-                )
-
-                yield loop_start_resp
-
-            elif isinstance(event, QueueLoopNextEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                loop_next_resp = self._workflow_response_converter.workflow_loop_next_to_stream_response(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_execution_id=self._workflow_run_id,
-                    event=event,
-                )
-
-                yield loop_next_resp
-
-            elif isinstance(event, QueueLoopCompletedEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-
-                loop_finish_resp = self._workflow_response_converter.workflow_loop_completed_to_stream_response(
-                    task_id=self._application_generate_entity.task_id,
-                    workflow_execution_id=self._workflow_run_id,
-                    event=event,
-                )
-
-                yield loop_finish_resp
-
-            elif isinstance(event, QueueWorkflowSucceededEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-                if not graph_runtime_state:
-                    raise ValueError("graph runtime state not initialized.")
-
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_success(
-                        workflow_run_id=self._workflow_run_id,
-                        total_tokens=graph_runtime_state.total_tokens,
-                        total_steps=graph_runtime_state.node_run_steps,
-                        outputs=event.outputs,
-                        conversation_id=None,
-                        trace_manager=trace_manager,
+                case QueueTextChunkEvent():
+                    yield from self._handle_text_chunk_event(
+                        event, tts_publisher=tts_publisher, queue_message=queue_message
                     )
 
-                    # save workflow app log
-                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
+                case QueueErrorEvent():
+                    yield from self._handle_error_event(event)
+                    break
 
-                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_execution=workflow_execution,
-                    )
-                    session.commit()
-
-                yield workflow_finish_resp
-            elif isinstance(event, QueueWorkflowPartialSuccessEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-                if not graph_runtime_state:
-                    raise ValueError("graph runtime state not initialized.")
-
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_partial_success(
-                        workflow_run_id=self._workflow_run_id,
-                        total_tokens=graph_runtime_state.total_tokens,
-                        total_steps=graph_runtime_state.node_run_steps,
-                        outputs=event.outputs,
-                        exceptions_count=event.exceptions_count,
-                        conversation_id=None,
-                        trace_manager=trace_manager,
-                    )
-
-                    # save workflow app log
-                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
-
-                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_execution=workflow_execution,
-                    )
-                    session.commit()
-
-                yield workflow_finish_resp
-            elif isinstance(event, QueueWorkflowFailedEvent | QueueStopEvent):
-                if not self._workflow_run_id:
-                    raise ValueError("workflow run not initialized.")
-                if not graph_runtime_state:
-                    raise ValueError("graph runtime state not initialized.")
-
-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_failed(
-                        workflow_run_id=self._workflow_run_id,
-                        total_tokens=graph_runtime_state.total_tokens,
-                        total_steps=graph_runtime_state.node_run_steps,
-                        status=WorkflowExecutionStatus.FAILED
-                        if isinstance(event, QueueWorkflowFailedEvent)
-                        else WorkflowExecutionStatus.STOPPED,
-                        error_message=event.error
-                        if isinstance(event, QueueWorkflowFailedEvent)
-                        else event.get_stop_reason(),
-                        conversation_id=None,
-                        trace_manager=trace_manager,
-                        exceptions_count=event.exceptions_count if isinstance(event, QueueWorkflowFailedEvent) else 0,
-                    )
-
-                    # save workflow app log
-                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)
-
-                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_execution=workflow_execution,
-                    )
-                    session.commit()
-
-                yield workflow_finish_resp
-            elif isinstance(event, QueueTextChunkEvent):
-                delta_text = event.text
-                if delta_text is None:
-                    continue
-
-                # only publish tts message at text chunk streaming
-                if tts_publisher:
-                    tts_publisher.publish(queue_message)
-
-                yield self._text_chunk_to_stream_response(
-                    delta_text, from_variable_selector=event.from_variable_selector
-                )
-            elif isinstance(event, QueueAgentLogEvent):
-                yield self._workflow_response_converter.handle_agent_log(
-                    task_id=self._application_generate_entity.task_id, event=event
-                )
-            else:
-                continue
+                # Handle all other events through elegant dispatch
+                case _:
+                    if responses := list(
+                        self._dispatch_event(
+                            event,
+                            graph_runtime_state=graph_runtime_state,
+                            tts_publisher=tts_publisher,
+                            trace_manager=trace_manager,
+                            queue_message=queue_message,
+                        )
+                    ):
+                        yield from responses
 
         if tts_publisher:
             tts_publisher.publish(None)

api/core/app/task_pipeline/exc.py

@@ -10,8 +10,3 @@ class RecordNotFoundError(TaskPipilineError):
 class WorkflowRunNotFoundError(RecordNotFoundError):
     def __init__(self, workflow_run_id: str):
         super().__init__("WorkflowRun", workflow_run_id)
-
-
-class WorkflowNodeExecutionNotFoundError(RecordNotFoundError):
-    def __init__(self, workflow_node_execution_id: str):
-        super().__init__("WorkflowNodeExecution", workflow_node_execution_id)

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]"
+    ANY = "any"
 
     # Dynamic select parameter
     # Once you are not sure about the available options until authorization is done

api/core/file/file_manager.py

@@ -7,6 +7,7 @@ from core.model_runtime.entities import (
     AudioPromptMessageContent,
     DocumentPromptMessageContent,
     ImagePromptMessageContent,
+    TextPromptMessageContent,
     VideoPromptMessageContent,
 )
 from core.model_runtime.entities.message_entities import PromptMessageContentUnionTypes
@@ -44,11 +45,44 @@ def to_prompt_message_content(
     *,
     image_detail_config: ImagePromptMessageContent.DETAIL | None = None,
 ) -> PromptMessageContentUnionTypes:
+    """
+    Convert a file to prompt message content.
+
+    This function converts files to their appropriate prompt message content types.
+    For supported file types (IMAGE, AUDIO, VIDEO, DOCUMENT), it creates the
+    corresponding message content with proper encoding/URL.
+
+    For unsupported file types, instead of raising an error, it returns a
+    TextPromptMessageContent with a descriptive message about the file.
+
+    Args:
+        f: The file to convert
+        image_detail_config: Optional detail configuration for image files
+
+    Returns:
+        PromptMessageContentUnionTypes: The appropriate message content type
+
+    Raises:
+        ValueError: If file extension or mime_type is missing
+    """
     if f.extension is None:
         raise ValueError("Missing file extension")
     if f.mime_type is None:
         raise ValueError("Missing file mime_type")
 
+    prompt_class_map: Mapping[FileType, type[PromptMessageContentUnionTypes]] = {
+        FileType.IMAGE: ImagePromptMessageContent,
+        FileType.AUDIO: AudioPromptMessageContent,
+        FileType.VIDEO: VideoPromptMessageContent,
+        FileType.DOCUMENT: DocumentPromptMessageContent,
+    }
+
+    # Check if file type is supported
+    if f.type not in prompt_class_map:
+        # For unsupported file types, return a text description
+        return TextPromptMessageContent(data=f"[Unsupported file type: {f.filename} ({f.type.value})]")
+
+    # Process supported file types
     params = {
         "base64_data": _get_encoded_string(f) if dify_config.MULTIMODAL_SEND_FORMAT == "base64" else "",
         "url": _to_url(f) if dify_config.MULTIMODAL_SEND_FORMAT == "url" else "",
@@ -58,17 +92,7 @@ def to_prompt_message_content(
     if f.type == FileType.IMAGE:
         params["detail"] = image_detail_config or ImagePromptMessageContent.DETAIL.LOW
 
-    prompt_class_map: Mapping[FileType, type[PromptMessageContentUnionTypes]] = {
-        FileType.IMAGE: ImagePromptMessageContent,
-        FileType.AUDIO: AudioPromptMessageContent,
-        FileType.VIDEO: VideoPromptMessageContent,
-        FileType.DOCUMENT: DocumentPromptMessageContent,
-    }
-
-    try:
-        return prompt_class_map[f.type].model_validate(params)
-    except KeyError:
-        raise ValueError(f"file type {f.type} is not supported")
+    return prompt_class_map[f.type].model_validate(params)
 
 
 def download(f: File, /):

api/core/file/tool_file_parser.py

@@ -7,13 +7,6 @@ if TYPE_CHECKING:
 _tool_file_manager_factory: Callable[[], "ToolFileManager"] | None = None
 
 
-class ToolFileParser:
-    @staticmethod
-    def get_tool_file_manager() -> "ToolFileManager":
-        assert _tool_file_manager_factory is not None
-        return _tool_file_manager_factory()
-
-
 def set_tool_file_manager_factory(factory: Callable[[], "ToolFileManager"]) -> None:
     global _tool_file_manager_factory
     _tool_file_manager_factory = factory

api/core/helper/encrypter.py

@@ -21,7 +21,7 @@ def encrypt_token(tenant_id: str, token: str):
     return base64.b64encode(encrypted_token).decode()
 
 
-def decrypt_token(tenant_id: str, token: str):
+def decrypt_token(tenant_id: str, token: str) -> str:
     return rsa.decrypt(base64.b64decode(token), tenant_id)
 
 

api/core/helper/provider_cache.py

@@ -0,0 +1,84 @@
+import json
+from abc import ABC, abstractmethod
+from json import JSONDecodeError
+from typing import Any, Optional
+
+from extensions.ext_redis import redis_client
+
+
+class ProviderCredentialsCache(ABC):
+    """Base class for provider credentials cache"""
+
+    def __init__(self, **kwargs):
+        self.cache_key = self._generate_cache_key(**kwargs)
+
+    @abstractmethod
+    def _generate_cache_key(self, **kwargs) -> str:
+        """Generate cache key based on subclass implementation"""
+        pass
+
+    def get(self) -> Optional[dict]:
+        """Get cached provider credentials"""
+        cached_credentials = redis_client.get(self.cache_key)
+        if cached_credentials:
+            try:
+                cached_credentials = cached_credentials.decode("utf-8")
+                return dict(json.loads(cached_credentials))
+            except JSONDecodeError:
+                return None
+        return None
+
+    def set(self, config: dict[str, Any]) -> None:
+        """Cache provider credentials"""
+        redis_client.setex(self.cache_key, 86400, json.dumps(config))
+
+    def delete(self) -> None:
+        """Delete cached provider credentials"""
+        redis_client.delete(self.cache_key)
+
+
+class SingletonProviderCredentialsCache(ProviderCredentialsCache):
+    """Cache for tool single provider credentials"""
+
+    def __init__(self, tenant_id: str, provider_type: str, provider_identity: str):
+        super().__init__(
+            tenant_id=tenant_id,
+            provider_type=provider_type,
+            provider_identity=provider_identity,
+        )
+
+    def _generate_cache_key(self, **kwargs) -> str:
+        tenant_id = kwargs["tenant_id"]
+        provider_type = kwargs["provider_type"]
+        identity_name = kwargs["provider_identity"]
+        identity_id = f"{provider_type}.{identity_name}"
+        return f"{provider_type}_credentials:tenant_id:{tenant_id}:id:{identity_id}"
+
+
+class ToolProviderCredentialsCache(ProviderCredentialsCache):
+    """Cache for tool provider credentials"""
+
+    def __init__(self, tenant_id: str, provider: str, credential_id: str):
+        super().__init__(tenant_id=tenant_id, provider=provider, credential_id=credential_id)
+
+    def _generate_cache_key(self, **kwargs) -> str:
+        tenant_id = kwargs["tenant_id"]
+        provider = kwargs["provider"]
+        credential_id = kwargs["credential_id"]
+        return f"tool_credentials:tenant_id:{tenant_id}:provider:{provider}:credential_id:{credential_id}"
+
+
+class NoOpProviderCredentialCache:
+    """No-op provider credential cache"""
+
+    def get(self) -> Optional[dict]:
+        """Get cached provider credentials"""
+        return None
+
+    def set(self, config: dict[str, Any]) -> None:
+        """Cache provider credentials"""
+        pass
+
+    def delete(self) -> None:
+        """Delete cached provider credentials"""
+        pass

api/core/helper/tool_provider_cache.py

@@ -1,51 +0,0 @@
-import json
-from enum import Enum
-from json import JSONDecodeError
-from typing import Optional
-
-from extensions.ext_redis import redis_client
-
-
-class ToolProviderCredentialsCacheType(Enum):
-    PROVIDER = "tool_provider"
-    ENDPOINT = "endpoint"
-
-
-class ToolProviderCredentialsCache:
-    def __init__(self, tenant_id: str, identity_id: str, cache_type: ToolProviderCredentialsCacheType):
-        self.cache_key = f"{cache_type.value}_credentials:tenant_id:{tenant_id}:id:{identity_id}"
-
-    def get(self) -> Optional[dict]:
-        """
-        Get cached model provider credentials.
-
-        :return:
-        """
-        cached_provider_credentials = redis_client.get(self.cache_key)
-        if cached_provider_credentials:
-            try:
-                cached_provider_credentials = cached_provider_credentials.decode("utf-8")
-                cached_provider_credentials = json.loads(cached_provider_credentials)
-            except JSONDecodeError:
-                return None
-
-            return dict(cached_provider_credentials)
-        else:
-            return None
-
-    def set(self, credentials: dict) -> None:
-        """
-        Cache model provider credentials.
-
-        :param credentials: provider credentials
-        :return:
-        """
-        redis_client.setex(self.cache_key, 86400, json.dumps(credentials))
-
-    def delete(self) -> None:
-        """
-        Delete cached model provider credentials.
-
-        :return:
-        """
-        redis_client.delete(self.cache_key)

api/core/helper/url_signer.py

@@ -1,52 +0,0 @@
-import base64
-import hashlib
-import hmac
-import os
-import time
-
-from pydantic import BaseModel, Field
-
-from configs import dify_config
-
-
-class SignedUrlParams(BaseModel):
-    sign_key: str = Field(..., description="The sign key")
-    timestamp: str = Field(..., description="Timestamp")
-    nonce: str = Field(..., description="Nonce")
-    sign: str = Field(..., description="Signature")
-
-
-class UrlSigner:
-    @classmethod
-    def get_signed_url(cls, url: str, sign_key: str, prefix: str) -> str:
-        signed_url_params = cls.get_signed_url_params(sign_key, prefix)
-        return (
-            f"{url}?timestamp={signed_url_params.timestamp}"
-            f"&nonce={signed_url_params.nonce}&sign={signed_url_params.sign}"
-        )
-
-    @classmethod
-    def get_signed_url_params(cls, sign_key: str, prefix: str) -> SignedUrlParams:
-        timestamp = str(int(time.time()))
-        nonce = os.urandom(16).hex()
-        sign = cls._sign(sign_key, timestamp, nonce, prefix)
-
-        return SignedUrlParams(sign_key=sign_key, timestamp=timestamp, nonce=nonce, sign=sign)
-
-    @classmethod
-    def verify(cls, sign_key: str, timestamp: str, nonce: str, sign: str, prefix: str) -> bool:
-        recalculated_sign = cls._sign(sign_key, timestamp, nonce, prefix)
-
-        return sign == recalculated_sign
-
-    @classmethod
-    def _sign(cls, sign_key: str, timestamp: str, nonce: str, prefix: str) -> str:
-        if not dify_config.SECRET_KEY:
-            raise Exception("SECRET_KEY is not set")
-
-        data_to_sign = f"{prefix}|{sign_key}|{timestamp}|{nonce}"
-        secret_key = dify_config.SECRET_KEY.encode()
-        sign = hmac.new(secret_key, data_to_sign.encode(), hashlib.sha256).digest()
-        encoded_sign = base64.urlsafe_b64encode(sign).decode()
-
-        return encoded_sign

api/core/llm_generator/llm_generator.py

@@ -148,9 +148,11 @@ class LLMGenerator:
 
             model_manager = ModelManager()
 
-            model_instance = model_manager.get_default_model_instance(
+            model_instance = model_manager.get_model_instance(
                 tenant_id=tenant_id,
                 model_type=ModelType.LLM,
+                provider=model_config.get("provider", ""),
+                model=model_config.get("name", ""),
             )
 
             try:

api/core/mcp/auth/auth_provider.py

@@ -8,7 +8,7 @@ from core.mcp.types import (
     OAuthTokens,
 )
 from models.tools import MCPToolProvider
-from services.tools.mcp_tools_mange_service import MCPToolManageService
+from services.tools.mcp_tools_manage_service import MCPToolManageService
 
 LATEST_PROTOCOL_VERSION = "1.0"
 

api/core/mcp/mcp_client.py

@@ -68,15 +68,17 @@ class MCPClient:
         }
 
         parsed_url = urlparse(self.server_url)
-        path = parsed_url.path
+        path = parsed_url.path or ""
         method_name = path.rstrip("/").split("/")[-1] if path else ""
-        try:
+        if method_name in connection_methods:
             client_factory = connection_methods[method_name]
             self.connect_server(client_factory, method_name)
-        except KeyError:
+        else:
             try:
+                logger.debug(f"Not supported method {method_name} found in URL path, trying default 'mcp' method.")
                 self.connect_server(sse_client, "sse")
             except MCPConnectionError:
+                logger.debug("MCP connection failed with 'sse', falling back to 'mcp' method.")
                 self.connect_server(streamablehttp_client, "mcp")
 
     def connect_server(
@@ -91,7 +93,7 @@ class MCPClient:
                 else {}
             )
             self._streams_context = client_factory(url=self.server_url, headers=headers)
-            if self._streams_context is None:
+            if not self._streams_context:
                 raise MCPConnectionError("Failed to create connection context")
 
             # Use exit_stack to manage context managers properly
@@ -141,10 +143,11 @@ class MCPClient:
         try:
             # ExitStack will handle proper cleanup of all managed context managers
             self.exit_stack.close()
+        except Exception as e:
+            logging.exception("Error during cleanup")
+            raise ValueError(f"Error during cleanup: {e}")
+        finally:
             self._session = None
             self._session_context = None
             self._streams_context = None
             self._initialized = False
-        except Exception as e:
-            logging.exception("Error during cleanup")
-            raise ValueError(f"Error during cleanup: {e}")

api/core/mcp/server/streamable_http.py

@@ -148,9 +148,7 @@ class MCPServerStreamableHTTPRequestHandler:
         if not self.end_user:
             raise ValueError("User not found")
         request = cast(types.CallToolRequest, self.request.root)
-        args = request.params.arguments
-        if not args:
-            raise ValueError("No arguments provided")
+        args = request.params.arguments or {}
         if self.app.mode in {AppMode.WORKFLOW.value}:
             args = {"inputs": args}
         elif self.app.mode in {AppMode.COMPLETION.value}:

api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -3,7 +3,7 @@ import json
 import logging
 import os
 from datetime import datetime, timedelta
-from typing import Optional, Union, cast
+from typing import Any, Optional, Union, cast
 
 from openinference.semconv.trace import OpenInferenceSpanKindValues, SpanAttributes
 from opentelemetry import trace
@@ -142,11 +142,8 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             raise
 
     def workflow_trace(self, trace_info: WorkflowTraceInfo):
-        if trace_info.message_data is None:
-            return
-
         workflow_metadata = {
-            "workflow_id": trace_info.workflow_run_id or "",
+            "workflow_run_id": trace_info.workflow_run_id or "",
             "message_id": trace_info.message_id or "",
             "workflow_app_log_id": trace_info.workflow_app_log_id or "",
             "status": trace_info.workflow_run_status or "",
@@ -156,7 +153,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
         }
         workflow_metadata.update(trace_info.metadata)
 
-        trace_id = uuid_to_trace_id(trace_info.message_id)
+        trace_id = uuid_to_trace_id(trace_info.workflow_run_id)
         span_id = RandomIdGenerator().generate_span_id()
         context = SpanContext(
             trace_id=trace_id,
@@ -213,7 +210,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                     if model:
                         node_metadata["ls_model_name"] = model
 
-                    outputs = json.loads(node_execution.outputs).get("usage", {})
+                    outputs = json.loads(node_execution.outputs).get("usage", {}) if "outputs" in node_execution else {}
                     usage_data = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
                     if usage_data:
                         node_metadata["total_tokens"] = usage_data.get("total_tokens", 0)
@@ -236,31 +233,34 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                         SpanAttributes.SESSION_ID: trace_info.conversation_id or "",
                     },
                     start_time=datetime_to_nanos(created_at),
+                    context=trace.set_span_in_context(trace.NonRecordingSpan(context)),
                 )
 
                 try:
                     if node_execution.node_type == "llm":
+                        llm_attributes: dict[str, Any] = {
+                            SpanAttributes.INPUT_VALUE: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
+                        }
                         provider = process_data.get("model_provider")
                         model = process_data.get("model_name")
                         if provider:
-                            node_span.set_attribute(SpanAttributes.LLM_PROVIDER, provider)
+                            llm_attributes[SpanAttributes.LLM_PROVIDER] = provider
                         if model:
-                            node_span.set_attribute(SpanAttributes.LLM_MODEL_NAME, model)
-
-                        outputs = json.loads(node_execution.outputs).get("usage", {})
+                            llm_attributes[SpanAttributes.LLM_MODEL_NAME] = model
+                        outputs = (
+                            json.loads(node_execution.outputs).get("usage", {}) if "outputs" in node_execution else {}
+                        )
                         usage_data = (
                             process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
                         )
                         if usage_data:
-                            node_span.set_attribute(
-                                SpanAttributes.LLM_TOKEN_COUNT_TOTAL, usage_data.get("total_tokens", 0)
-                            )
-                            node_span.set_attribute(
-                                SpanAttributes.LLM_TOKEN_COUNT_PROMPT, usage_data.get("prompt_tokens", 0)
-                            )
-                            node_span.set_attribute(
-                                SpanAttributes.LLM_TOKEN_COUNT_COMPLETION, usage_data.get("completion_tokens", 0)
+                            llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_TOTAL] = usage_data.get("total_tokens", 0)
+                            llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_PROMPT] = usage_data.get("prompt_tokens", 0)
+                            llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_COMPLETION] = usage_data.get(
+                                "completion_tokens", 0
                             )
+                        llm_attributes.update(self._construct_llm_attributes(process_data.get("prompts", [])))
+                        node_span.set_attributes(llm_attributes)
                 finally:
                     node_span.end(end_time=datetime_to_nanos(finished_at))
         finally:
@@ -352,25 +352,7 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
                 SpanAttributes.METADATA: json.dumps(message_metadata, ensure_ascii=False),
                 SpanAttributes.SESSION_ID: trace_info.message_data.conversation_id,
             }
-
-            if isinstance(trace_info.inputs, list):
-                for i, msg in enumerate(trace_info.inputs):
-                    if isinstance(msg, dict):
-                        llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.content"] = msg.get("text", "")
-                        llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.role"] = msg.get(
-                            "role", "user"
-                        )
-                        # todo: handle assistant and tool role messages, as they don't always
-                        # have a text field, but may have a tool_calls field instead
-                        # e.g. 'tool_calls': [{'id': '98af3a29-b066-45a5-b4b1-46c74ddafc58',
-                        # 'type': 'function', 'function': {'name': 'current_time', 'arguments': '{}'}}]}
-            elif isinstance(trace_info.inputs, dict):
-                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = json.dumps(trace_info.inputs)
-                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
-            elif isinstance(trace_info.inputs, str):
-                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = trace_info.inputs
-                llm_attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
-
+            llm_attributes.update(self._construct_llm_attributes(trace_info.inputs))
             if trace_info.total_tokens is not None and trace_info.total_tokens > 0:
                 llm_attributes[SpanAttributes.LLM_TOKEN_COUNT_TOTAL] = trace_info.total_tokens
             if trace_info.message_tokens is not None and trace_info.message_tokens > 0:
@@ -724,3 +706,24 @@ class ArizePhoenixDataTrace(BaseTraceInstance):
             .all()
         )
         return workflow_nodes
+
+    def _construct_llm_attributes(self, prompts: dict | list | str | None) -> dict[str, str]:
+        """Helper method to construct LLM attributes with passed prompts."""
+        attributes = {}
+        if isinstance(prompts, list):
+            for i, msg in enumerate(prompts):
+                if isinstance(msg, dict):
+                    attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.content"] = msg.get("text", "")
+                    attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.{i}.message.role"] = msg.get("role", "user")
+                    # todo: handle assistant and tool role messages, as they don't always
+                    # have a text field, but may have a tool_calls field instead
+                    # e.g. 'tool_calls': [{'id': '98af3a29-b066-45a5-b4b1-46c74ddafc58',
+                    # 'type': 'function', 'function': {'name': 'current_time', 'arguments': '{}'}}]}
+        elif isinstance(prompts, dict):
+            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = json.dumps(prompts)
+            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
+        elif isinstance(prompts, str):
+            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.content"] = prompts
+            attributes[f"{SpanAttributes.LLM_INPUT_MESSAGES}.0.message.role"] = "user"
+
+        return attributes

api/core/plugin/backwards_invocation/encrypt.py

@@ -1,16 +1,20 @@
+from core.helper.provider_cache import SingletonProviderCredentialsCache
 from core.plugin.entities.request import RequestInvokeEncrypt
-from core.tools.utils.configuration import ProviderConfigEncrypter
+from core.tools.utils.encryption import create_provider_encrypter
 from models.account import Tenant
 
 
 class PluginEncrypter:
     @classmethod
     def invoke_encrypt(cls, tenant: Tenant, payload: RequestInvokeEncrypt) -> dict:
-        encrypter = ProviderConfigEncrypter(
+        encrypter, cache = create_provider_encrypter(
             tenant_id=tenant.id,
             config=payload.config,
-            provider_type=payload.namespace,
-            provider_identity=payload.identity,
+            cache=SingletonProviderCredentialsCache(
+                tenant_id=tenant.id,
+                provider_type=payload.namespace,
+                provider_identity=payload.identity,
+            ),
         )
 
         if payload.opt == "encrypt":
@@ -22,7 +26,7 @@ class PluginEncrypter:
                 "data": encrypter.decrypt(payload.data),
             }
         elif payload.opt == "clear":
-            encrypter.delete_tool_credentials_cache()
+            cache.delete()
             return {
                 "data": {},
             }

api/core/plugin/backwards_invocation/tool.py

@@ -1,5 +1,5 @@
 from collections.abc import Generator
-from typing import Any
+from typing import Any, Optional
 
 from core.callback_handler.workflow_tool_callback_handler import DifyWorkflowCallbackHandler
 from core.plugin.backwards_invocation.base import BaseBackwardsInvocation
@@ -23,6 +23,7 @@ class PluginToolBackwardsInvocation(BaseBackwardsInvocation):
         provider: str,
         tool_name: str,
         tool_parameters: dict[str, Any],
+        credential_id: Optional[str] = None,
     ) -> Generator[ToolInvokeMessage, None, None]:
         """
         invoke tool
@@ -30,7 +31,7 @@ class PluginToolBackwardsInvocation(BaseBackwardsInvocation):
         # get tool runtime
         try:
             tool_runtime = ToolManager.get_tool_runtime_from_plugin(
-                tool_type, tenant_id, provider, tool_name, tool_parameters
+                tool_type, tenant_id, provider, tool_name, tool_parameters, credential_id
             )
             response = ToolEngine.generic_invoke(
                 tool_runtime, tool_parameters, user_id, DifyWorkflowCallbackHandler(), workflow_call_depth=1

api/core/plugin/entities/marketplace.py

@@ -32,6 +32,13 @@ class MarketplacePluginDeclaration(BaseModel):
     latest_package_identifier: str = Field(
         ..., description="Unique identifier for the latest package release of the plugin"
     )
+    status: str = Field(..., description="Indicate the status of marketplace plugin, enum from `active` `deleted`")
+    deprecated_reason: str = Field(
+        ..., description="Not empty when status='deleted', indicates the reason why this plugin is deleted(deprecated)"
+    )
+    alternative_plugin_id: str = Field(
+        ..., description="Optional, indicates the alternative plugin for user to switch to"
+    )
 
     @model_validator(mode="before")
     @classmethod

api/core/plugin/entities/parameters.py

@@ -5,6 +5,7 @@ from pydantic import BaseModel, Field, field_validator
 
 from core.entities.parameter_entities import CommonParameterType
 from core.tools.entities.common_entities import I18nObject
+from core.workflow.nodes.base.entities import NumberType
 
 
 class PluginParameterOption(BaseModel):
@@ -38,6 +39,7 @@ class PluginParameterType(enum.StrEnum):
     APP_SELECTOR = CommonParameterType.APP_SELECTOR.value
     MODEL_SELECTOR = CommonParameterType.MODEL_SELECTOR.value
     TOOLS_SELECTOR = CommonParameterType.TOOLS_SELECTOR.value
+    ANY = CommonParameterType.ANY.value
     DYNAMIC_SELECT = CommonParameterType.DYNAMIC_SELECT.value
 
     # deprecated, should not use.
@@ -151,6 +153,10 @@ def cast_parameter_value(typ: enum.StrEnum, value: Any, /):
                 if value and not isinstance(value, list):
                     raise ValueError("The tools selector must be a list.")
                 return value
+            case PluginParameterType.ANY:
+                if value and not isinstance(value, str | dict | list | NumberType):
+                    raise ValueError("The var selector must be a string, dictionary, list or number.")
+                return value
             case PluginParameterType.ARRAY:
                 if not isinstance(value, list):
                     # Try to parse JSON string for arrays

api/core/plugin/entities/plugin.py

@@ -135,17 +135,6 @@ class PluginEntity(PluginInstallation):
         return self
 
 
-class GithubPackage(BaseModel):
-    repo: str
-    version: str
-    package: str
-
-
-class GithubVersion(BaseModel):
-    repo: str
-    version: str
-
-
 class GenericProviderID:
     organization: str
     plugin_name: str

api/core/plugin/entities/plugin_daemon.py

@@ -182,6 +182,10 @@ class PluginOAuthAuthorizationUrlResponse(BaseModel):
 
 
 class PluginOAuthCredentialsResponse(BaseModel):
+    metadata: Mapping[str, Any] = Field(
+        default_factory=dict, description="The metadata of the OAuth, like avatar url, name, etc."
+    )
+    expires_at: int = Field(default=-1, description="The expires at time of the credentials. UTC timestamp.")
     credentials: Mapping[str, Any] = Field(description="The credentials of the OAuth.")
 
 

api/core/plugin/entities/request.py

@@ -27,6 +27,20 @@ from core.workflow.nodes.question_classifier.entities import (
 )
 
 
+class InvokeCredentials(BaseModel):
+    tool_credentials: dict[str, str] = Field(
+        default_factory=dict,
+        description="Map of tool provider to credential id, used to store the credential id for the tool provider.",
+    )
+
+
+class PluginInvokeContext(BaseModel):
+    credentials: Optional[InvokeCredentials] = Field(
+        default_factory=InvokeCredentials,
+        description="Credentials context for the plugin invocation or backward invocation.",
+    )
+
+
 class RequestInvokeTool(BaseModel):
     """
     Request to invoke a tool
@@ -36,6 +50,7 @@ class RequestInvokeTool(BaseModel):
     provider: str
     tool: str
     tool_parameters: dict
+    credential_id: Optional[str] = None
 
 
 class BaseRequestInvokeModel(BaseModel):

api/core/plugin/impl/agent.py

@@ -6,6 +6,7 @@ from core.plugin.entities.plugin import GenericProviderID
 from core.plugin.entities.plugin_daemon import (
     PluginAgentProviderEntity,
 )
+from core.plugin.entities.request import PluginInvokeContext
 from core.plugin.impl.base import BasePluginClient
 
 
@@ -83,6 +84,7 @@ class PluginAgentClient(BasePluginClient):
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
         message_id: Optional[str] = None,
+        context: Optional[PluginInvokeContext] = None,
     ) -> Generator[AgentInvokeMessage, None, None]:
         """
         Invoke the agent with the given tenant, user, plugin, provider, name and parameters.
@@ -99,6 +101,7 @@ class PluginAgentClient(BasePluginClient):
                 "conversation_id": conversation_id,
                 "app_id": app_id,
                 "message_id": message_id,
+                "context": context.model_dump() if context else {},
                 "data": {
                     "agent_strategy_provider": agent_provider_id.provider_name,
                     "agent_strategy": agent_strategy,

api/core/plugin/impl/oauth.py

@@ -15,27 +15,32 @@ class OAuthHandler(BasePluginClient):
         user_id: str,
         plugin_id: str,
         provider: str,
+        redirect_uri: str,
         system_credentials: Mapping[str, Any],
     ) -> PluginOAuthAuthorizationUrlResponse:
-        response = self._request_with_plugin_daemon_response_stream(
-            "POST",
-            f"plugin/{tenant_id}/dispatch/oauth/get_authorization_url",
-            PluginOAuthAuthorizationUrlResponse,
-            data={
-                "user_id": user_id,
-                "data": {
-                    "provider": provider,
-                    "system_credentials": system_credentials,
+        try:
+            response = self._request_with_plugin_daemon_response_stream(
+                "POST",
+                f"plugin/{tenant_id}/dispatch/oauth/get_authorization_url",
+                PluginOAuthAuthorizationUrlResponse,
+                data={
+                    "user_id": user_id,
+                    "data": {
+                        "provider": provider,
+                        "redirect_uri": redirect_uri,
+                        "system_credentials": system_credentials,
+                    },
                 },
-            },
-            headers={
-                "X-Plugin-ID": plugin_id,
-                "Content-Type": "application/json",
-            },
-        )
-        for resp in response:
-            return resp
-        raise ValueError("No response received from plugin daemon for authorization URL request.")
+                headers={
+                    "X-Plugin-ID": plugin_id,
+                    "Content-Type": "application/json",
+                },
+            )
+            for resp in response:
+                return resp
+            raise ValueError("No response received from plugin daemon for authorization URL request.")
+        except Exception as e:
+            raise ValueError(f"Error getting authorization URL: {e}")
 
     def get_credentials(
         self,
@@ -43,6 +48,7 @@ class OAuthHandler(BasePluginClient):
         user_id: str,
         plugin_id: str,
         provider: str,
+        redirect_uri: str,
         system_credentials: Mapping[str, Any],
         request: Request,
     ) -> PluginOAuthCredentialsResponse:
@@ -50,30 +56,68 @@ class OAuthHandler(BasePluginClient):
         Get credentials from the given request.
         """
 
-        # encode request to raw http request
-        raw_request_bytes = self._convert_request_to_raw_data(request)
-
-        response = self._request_with_plugin_daemon_response_stream(
-            "POST",
-            f"plugin/{tenant_id}/dispatch/oauth/get_credentials",
-            PluginOAuthCredentialsResponse,
-            data={
-                "user_id": user_id,
-                "data": {
-                    "provider": provider,
-                    "system_credentials": system_credentials,
-                    # for json serialization
-                    "raw_http_request": binascii.hexlify(raw_request_bytes).decode(),
+        try:
+            # encode request to raw http request
+            raw_request_bytes = self._convert_request_to_raw_data(request)
+            response = self._request_with_plugin_daemon_response_stream(
+                "POST",
+                f"plugin/{tenant_id}/dispatch/oauth/get_credentials",
+                PluginOAuthCredentialsResponse,
+                data={
+                    "user_id": user_id,
+                    "data": {
+                        "provider": provider,
+                        "redirect_uri": redirect_uri,
+                        "system_credentials": system_credentials,
+                        # for json serialization
+                        "raw_http_request": binascii.hexlify(raw_request_bytes).decode(),
+                    },
                 },
-            },
-            headers={
-                "X-Plugin-ID": plugin_id,
-                "Content-Type": "application/json",
-            },
-        )
-        for resp in response:
-            return resp
-        raise ValueError("No response received from plugin daemon for authorization URL request.")
+                headers={
+                    "X-Plugin-ID": plugin_id,
+                    "Content-Type": "application/json",
+                },
+            )
+            for resp in response:
+                return resp
+            raise ValueError("No response received from plugin daemon for authorization URL request.")
+        except Exception as e:
+            raise ValueError(f"Error getting credentials: {e}")
+
+    def refresh_credentials(
+        self,
+        tenant_id: str,
+        user_id: str,
+        plugin_id: str,
+        provider: str,
+        redirect_uri: str,
+        system_credentials: Mapping[str, Any],
+        credentials: Mapping[str, Any],
+    ) -> PluginOAuthCredentialsResponse:
+        try:
+            response = self._request_with_plugin_daemon_response_stream(
+                "POST",
+                f"plugin/{tenant_id}/dispatch/oauth/refresh_credentials",
+                PluginOAuthCredentialsResponse,
+                data={
+                    "user_id": user_id,
+                    "data": {
+                        "provider": provider,
+                        "redirect_uri": redirect_uri,
+                        "system_credentials": system_credentials,
+                        "credentials": credentials,
+                    },
+                },
+                headers={
+                    "X-Plugin-ID": plugin_id,
+                    "Content-Type": "application/json",
+                },
+            )
+            for resp in response:
+                return resp
+            raise ValueError("No response received from plugin daemon for refresh credentials request.")
+        except Exception as e:
+            raise ValueError(f"Error refreshing credentials: {e}")
 
     def _convert_request_to_raw_data(self, request: Request) -> bytes:
         """

api/core/plugin/impl/tool.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel
 from core.plugin.entities.plugin import GenericProviderID, ToolProviderID
 from core.plugin.entities.plugin_daemon import PluginBasicBooleanResponse, PluginToolProviderEntity
 from core.plugin.impl.base import BasePluginClient
-from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
+from core.tools.entities.tool_entities import CredentialType, ToolInvokeMessage, ToolParameter
 
 
 class PluginToolManager(BasePluginClient):
@@ -78,6 +78,7 @@ class PluginToolManager(BasePluginClient):
         tool_provider: str,
         tool_name: str,
         credentials: dict[str, Any],
+        credential_type: CredentialType,
         tool_parameters: dict[str, Any],
         conversation_id: Optional[str] = None,
         app_id: Optional[str] = None,
@@ -102,6 +103,7 @@ class PluginToolManager(BasePluginClient):
                     "provider": tool_provider_id.provider_name,
                     "tool": tool_name,
                     "credentials": credentials,
+                    "credential_type": credential_type,
                     "tool_parameters": tool_parameters,
                 },
             },

api/core/prompt/simple_prompt_transform.py

@@ -29,19 +29,6 @@ class ModelMode(enum.StrEnum):
     COMPLETION = "completion"
     CHAT = "chat"
 
-    @classmethod
-    def value_of(cls, value: str) -> "ModelMode":
-        """
-        Get value of given mode.
-
-        :param value: mode value
-        :return: mode
-        """
-        for mode in cls:
-            if mode.value == value:
-                return mode
-        raise ValueError(f"invalid mode value {value}")
-
 
 prompt_file_contents: dict[str, Any] = {}
 
@@ -65,7 +52,7 @@ class SimplePromptTransform(PromptTransform):
     ) -> tuple[list[PromptMessage], Optional[list[str]]]:
         inputs = {key: str(value) for key, value in inputs.items()}
 
-        model_mode = ModelMode.value_of(model_config.mode)
+        model_mode = ModelMode(model_config.mode)
         if model_mode == ModelMode.CHAT:
             prompt_messages, stops = self._get_chat_model_prompt_messages(
                 app_mode=app_mode,

api/core/rag/cleaner/unstructured/unstructured_extra_whitespace_cleaner.py

@@ -1,12 +0,0 @@
-"""Abstract interface for document clean implementations."""
-
-from core.rag.cleaner.cleaner_base import BaseCleaner
-
-
-class UnstructuredNonAsciiCharsCleaner(BaseCleaner):
-    def clean(self, content) -> str:
-        """clean document content."""
-        from unstructured.cleaners.core import clean_extra_whitespace
-
-        # Returns "ITEM 1A: RISK FACTORS"
-        return clean_extra_whitespace(content)

api/core/rag/cleaner/unstructured/unstructured_group_broken_paragraphs_cleaner.py

@@ -1,15 +0,0 @@
-"""Abstract interface for document clean implementations."""
-
-from core.rag.cleaner.cleaner_base import BaseCleaner
-
-
-class UnstructuredGroupBrokenParagraphsCleaner(BaseCleaner):
-    def clean(self, content) -> str:
-        """clean document content."""
-        import re
-
-        from unstructured.cleaners.core import group_broken_paragraphs
-
-        para_split_re = re.compile(r"(\s*\n\s*){3}")
-
-        return group_broken_paragraphs(content, paragraph_split=para_split_re)

api/core/rag/cleaner/unstructured/unstructured_non_ascii_chars_cleaner.py

@@ -1,12 +0,0 @@
-"""Abstract interface for document clean implementations."""
-
-from core.rag.cleaner.cleaner_base import BaseCleaner
-
-
-class UnstructuredNonAsciiCharsCleaner(BaseCleaner):
-    def clean(self, content) -> str:
-        """clean document content."""
-        from unstructured.cleaners.core import clean_non_ascii_chars
-
-        # Returns "This text contains non-ascii characters!"
-        return clean_non_ascii_chars(content)

api/core/rag/cleaner/unstructured/unstructured_replace_unicode_quotes_cleaner.py

@@ -1,12 +0,0 @@
-"""Abstract interface for document clean implementations."""
-
-from core.rag.cleaner.cleaner_base import BaseCleaner
-
-
-class UnstructuredNonAsciiCharsCleaner(BaseCleaner):
-    def clean(self, content) -> str:
-        """Replaces unicode quote characters, such as the \x91 character in a string."""
-
-        from unstructured.cleaners.core import replace_unicode_quotes
-
-        return replace_unicode_quotes(content)

api/core/rag/cleaner/unstructured/unstructured_translate_text_cleaner.py

@@ -1,11 +0,0 @@
-"""Abstract interface for document clean implementations."""
-
-from core.rag.cleaner.cleaner_base import BaseCleaner
-
-
-class UnstructuredTranslateTextCleaner(BaseCleaner):
-    def clean(self, content) -> str:
-        """clean document content."""
-        from unstructured.cleaners.translate import translate_text
-
-        return translate_text(content)

api/core/rag/datasource/vdb/analyticdb/analyticdb_vector_openapi.py

@@ -233,6 +233,12 @@ class AnalyticdbVectorOpenAPI:
     def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:
         from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
 
+        document_ids_filter = kwargs.get("document_ids_filter")
+        where_clause = ""
+        if document_ids_filter:
+            document_ids = ", ".join(f"'{id}'" for id in document_ids_filter)
+            where_clause += f"metadata_->>'document_id' IN ({document_ids})"
+
         score_threshold = kwargs.get("score_threshold") or 0.0
         request = gpdb_20160503_models.QueryCollectionDataRequest(
             dbinstance_id=self.config.instance_id,
@@ -245,7 +251,7 @@ class AnalyticdbVectorOpenAPI:
             vector=query_vector,
             content=None,
             top_k=kwargs.get("top_k", 4),
-            filter=None,
+            filter=where_clause,
         )
         response = self._client.query_collection_data(request)
         documents = []
@@ -265,6 +271,11 @@ class AnalyticdbVectorOpenAPI:
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
         from alibabacloud_gpdb20160503 import models as gpdb_20160503_models
 
+        document_ids_filter = kwargs.get("document_ids_filter")
+        where_clause = ""
+        if document_ids_filter:
+            document_ids = ", ".join(f"'{id}'" for id in document_ids_filter)
+            where_clause += f"metadata_->>'document_id' IN ({document_ids})"
         score_threshold = float(kwargs.get("score_threshold") or 0.0)
         request = gpdb_20160503_models.QueryCollectionDataRequest(
             dbinstance_id=self.config.instance_id,
@@ -277,7 +288,7 @@ class AnalyticdbVectorOpenAPI:
             vector=None,
             content=query,
             top_k=kwargs.get("top_k", 4),
-            filter=None,
+            filter=where_clause,
         )
         response = self._client.query_collection_data(request)
         documents = []

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

@@ -147,10 +147,17 @@ class ElasticSearchVector(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: dict[str, Any] = {"match": {Field.CONTENT_KEY.value: query}}
         document_ids_filter = kwargs.get("document_ids_filter")
+
         if document_ids_filter:
-            query_str["filter"] = {"terms": {"metadata.document_id": document_ids_filter}}  # type: ignore
+            query_str = {
+                "bool": {
+                    "must": {"match": {Field.CONTENT_KEY.value: query}},
+                    "filter": {"terms": {"metadata.document_id": document_ids_filter}},
+                }
+            }
+
         results = self._client.search(index=self._collection_name, query=query_str, size=kwargs.get("top_k", 4))
         docs = []
         for hit in results["hits"]["hits"]:

api/core/rag/datasource/vdb/tencent/tencent_vector.py

@@ -206,9 +206,19 @@ class TencentVector(BaseVector):
     def delete_by_ids(self, ids: list[str]) -> None:
         if not ids:
             return
-        self._client.delete(
-            database_name=self._client_config.database, collection_name=self.collection_name, document_ids=ids
-        )
+
+        total_count = len(ids)
+        batch_size = self._client_config.max_upsert_batch_size
+        batch = math.ceil(total_count / batch_size)
+
+        for j in range(batch):
+            start_idx = j * batch_size
+            end_idx = min(total_count, (j + 1) * batch_size)
+            batch_ids = ids[start_idx:end_idx]
+
+            self._client.delete(
+                database_name=self._client_config.database, collection_name=self.collection_name, document_ids=batch_ids
+            )
 
     def delete_by_metadata_field(self, key: str, value: str) -> None:
         self._client.delete(

api/core/rag/datasource/vdb/tidb_on_qdrant/tidb_entities.py

@@ -1,17 +0,0 @@
-from typing import Optional
-
-from pydantic import BaseModel
-
-
-class ClusterEntity(BaseModel):
-    """
-    Model Config Entity.
-    """
-
-    name: str
-    cluster_id: str
-    displayName: str
-    region: str
-    spendingLimit: Optional[int] = 1000
-    version: str
-    createdBy: str

api/core/rag/extractor/blob/blob.py

@@ -9,8 +9,7 @@ from __future__ import annotations
 
 import contextlib
 import mimetypes
-from abc import ABC, abstractmethod
-from collections.abc import Generator, Iterable, Mapping
+from collections.abc import Generator, Mapping
 from io import BufferedReader, BytesIO
 from pathlib import Path, PurePath
 from typing import Any, Optional, Union
@@ -143,21 +142,3 @@ class Blob(BaseModel):
         if self.source:
             str_repr += f" {self.source}"
         return str_repr
-
-
-class BlobLoader(ABC):
-    """Abstract interface for blob loaders implementation.
-
-    Implementer should be able to load raw content from a datasource system according
-    to some criteria and return the raw content lazily as a stream of blobs.
-    """
-
-    @abstractmethod
-    def yield_blobs(
-        self,
-    ) -> Iterable[Blob]:
-        """A lazy loader for raw data represented by Blob object.
-
-        Returns:
-            A generator over blobs
-        """

api/core/rag/extractor/unstructured/unstructured_pdf_extractor.py

@@ -1,47 +0,0 @@
-import logging
-
-from core.rag.extractor.extractor_base import BaseExtractor
-from core.rag.models.document import Document
-
-logger = logging.getLogger(__name__)
-
-
-class UnstructuredPDFExtractor(BaseExtractor):
-    """Load pdf files.
-
-
-    Args:
-        file_path: Path to the file to load.
-
-        api_url: Unstructured API URL
-
-        api_key: Unstructured API Key
-    """
-
-    def __init__(self, file_path: str, api_url: str, api_key: str):
-        """Initialize with file path."""
-        self._file_path = file_path
-        self._api_url = api_url
-        self._api_key = api_key
-
-    def extract(self) -> list[Document]:
-        if self._api_url:
-            from unstructured.partition.api import partition_via_api
-
-            elements = partition_via_api(
-                filename=self._file_path, api_url=self._api_url, api_key=self._api_key, strategy="auto"
-            )
-        else:
-            from unstructured.partition.pdf import partition_pdf
-
-            elements = partition_pdf(filename=self._file_path, strategy="auto")
-
-        from unstructured.chunking.title import chunk_by_title
-
-        chunks = chunk_by_title(elements, max_characters=2000, combine_text_under_n_chars=2000)
-        documents = []
-        for chunk in chunks:
-            text = chunk.text.strip()
-            documents.append(Document(page_content=text))
-
-        return documents

api/core/rag/extractor/unstructured/unstructured_text_extractor.py

@@ -1,34 +0,0 @@
-import logging
-
-from core.rag.extractor.extractor_base import BaseExtractor
-from core.rag.models.document import Document
-
-logger = logging.getLogger(__name__)
-
-
-class UnstructuredTextExtractor(BaseExtractor):
-    """Load msg files.
-
-
-    Args:
-        file_path: Path to the file to load.
-    """
-
-    def __init__(self, file_path: str, api_url: str):
-        """Initialize with file path."""
-        self._file_path = file_path
-        self._api_url = api_url
-
-    def extract(self) -> list[Document]:
-        from unstructured.partition.text import partition_text
-
-        elements = partition_text(filename=self._file_path)
-        from unstructured.chunking.title import chunk_by_title
-
-        chunks = chunk_by_title(elements, max_characters=2000, combine_text_under_n_chars=2000)
-        documents = []
-        for chunk in chunks:
-            text = chunk.text.strip()
-            documents.append(Document(page_content=text))
-
-        return documents

api/core/rag/extractor/word_extractor.py

@@ -238,9 +238,11 @@ class WordExtractor(BaseExtractor):
             paragraph_content = []
             for run in paragraph.runs:
                 if hasattr(run.element, "tag") and isinstance(run.element.tag, str) and run.element.tag.endswith("r"):
+                    # Process drawing type images
                     drawing_elements = run.element.findall(
                         ".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}drawing"
                     )
+                    has_drawing = False
                     for drawing in drawing_elements:
                         blip_elements = drawing.findall(
                             ".//{http://schemas.openxmlformats.org/drawingml/2006/main}blip"
@@ -252,6 +254,34 @@ class WordExtractor(BaseExtractor):
                             if embed_id:
                                 image_part = doc.part.related_parts.get(embed_id)
                                 if image_part in image_map:
+                                    has_drawing = True
+                                    paragraph_content.append(image_map[image_part])
+                    # Process pict type images
+                    shape_elements = run.element.findall(
+                        ".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}pict"
+                    )
+                    for shape in shape_elements:
+                        # Find image data in VML
+                        shape_image = shape.find(
+                            ".//{http://schemas.openxmlformats.org/wordprocessingml/2006/main}binData"
+                        )
+                        if shape_image is not None and shape_image.text:
+                            image_id = shape_image.get(
+                                "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
+                            )
+                            if image_id and image_id in doc.part.rels:
+                                image_part = doc.part.rels[image_id].target_part
+                                if image_part in image_map and not has_drawing:
+                                    paragraph_content.append(image_map[image_part])
+                        # Find imagedata element in VML
+                        image_data = shape.find(".//{urn:schemas-microsoft-com:vml}imagedata")
+                        if image_data is not None:
+                            image_id = image_data.get("id") or image_data.get(
+                                "{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"
+                            )
+                            if image_id and image_id in doc.part.rels:
+                                image_part = doc.part.rels[image_id].target_part
+                                if image_part in image_map and not has_drawing:
                                     paragraph_content.append(image_map[image_part])
                 if run.text.strip():
                     paragraph_content.append(run.text.strip())

api/core/rag/retrieval/dataset_retrieval.py

@@ -1137,7 +1137,7 @@ class DatasetRetrieval:
     def _get_prompt_template(
         self, model_config: ModelConfigWithCredentialsEntity, mode: str, metadata_fields: list, query: str
     ):
-        model_mode = ModelMode.value_of(mode)
+        model_mode = ModelMode(mode)
         input_text = query
 
         prompt_template: Union[CompletionModelPromptTemplate, list[ChatModelMessage]]

api/core/rag/splitter/fixed_text_splitter.py

@@ -102,6 +102,7 @@ class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter)
                 splits = text.split()
             else:
                 splits = text.split(separator)
+                splits = [item + separator if i < len(splits) else item for i, item in enumerate(splits)]
         else:
             splits = list(text)
         splits = [s for s in splits if (s not in {"", "\n"})]

api/core/rag/splitter/text_splitter.py

@@ -10,7 +10,6 @@ from typing import (
     Any,
     Literal,
     Optional,
-    TypedDict,
     TypeVar,
     Union,
 )
@@ -168,167 +167,6 @@ class TextSplitter(BaseDocumentTransformer, ABC):
         raise NotImplementedError
 
 
-class CharacterTextSplitter(TextSplitter):
-    """Splitting text that looks at characters."""
-
-    def __init__(self, separator: str = "\n\n", **kwargs: Any) -> None:
-        """Create a new TextSplitter."""
-        super().__init__(**kwargs)
-        self._separator = separator
-
-    def split_text(self, text: str) -> list[str]:
-        """Split incoming text and return chunks."""
-        # First we naively split the large input into a bunch of smaller ones.
-        splits = _split_text_with_regex(text, self._separator, self._keep_separator)
-        _separator = "" if self._keep_separator else self._separator
-        _good_splits_lengths = []  # cache the lengths of the splits
-        if splits:
-            _good_splits_lengths.extend(self._length_function(splits))
-        return self._merge_splits(splits, _separator, _good_splits_lengths)
-
-
-class LineType(TypedDict):
-    """Line type as typed dict."""
-
-    metadata: dict[str, str]
-    content: str
-
-
-class HeaderType(TypedDict):
-    """Header type as typed dict."""
-
-    level: int
-    name: str
-    data: str
-
-
-class MarkdownHeaderTextSplitter:
-    """Splitting markdown files based on specified headers."""
-
-    def __init__(self, headers_to_split_on: list[tuple[str, str]], return_each_line: bool = False):
-        """Create a new MarkdownHeaderTextSplitter.
-
-        Args:
-            headers_to_split_on: Headers we want to track
-            return_each_line: Return each line w/ associated headers
-        """
-        # Output line-by-line or aggregated into chunks w/ common headers
-        self.return_each_line = return_each_line
-        # Given the headers we want to split on,
-        # (e.g., "#, ##, etc") order by length
-        self.headers_to_split_on = sorted(headers_to_split_on, key=lambda split: len(split[0]), reverse=True)
-
-    def aggregate_lines_to_chunks(self, lines: list[LineType]) -> list[Document]:
-        """Combine lines with common metadata into chunks
-        Args:
-            lines: Line of text / associated header metadata
-        """
-        aggregated_chunks: list[LineType] = []
-
-        for line in lines:
-            if aggregated_chunks and aggregated_chunks[-1]["metadata"] == line["metadata"]:
-                # If the last line in the aggregated list
-                # has the same metadata as the current line,
-                # append the current content to the last lines's content
-                aggregated_chunks[-1]["content"] += "  \n" + line["content"]
-            else:
-                # Otherwise, append the current line to the aggregated list
-                aggregated_chunks.append(line)
-
-        return [Document(page_content=chunk["content"], metadata=chunk["metadata"]) for chunk in aggregated_chunks]
-
-    def split_text(self, text: str) -> list[Document]:
-        """Split markdown file
-        Args:
-            text: Markdown file"""
-
-        # Split the input text by newline character ("\n").
-        lines = text.split("\n")
-        # Final output
-        lines_with_metadata: list[LineType] = []
-        # Content and metadata of the chunk currently being processed
-        current_content: list[str] = []
-        current_metadata: dict[str, str] = {}
-        # Keep track of the nested header structure
-        # header_stack: List[Dict[str, Union[int, str]]] = []
-        header_stack: list[HeaderType] = []
-        initial_metadata: dict[str, str] = {}
-
-        for line in lines:
-            stripped_line = line.strip()
-            # Check each line against each of the header types (e.g., #, ##)
-            for sep, name in self.headers_to_split_on:
-                # Check if line starts with a header that we intend to split on
-                if stripped_line.startswith(sep) and (
-                    # Header with no text OR header is followed by space
-                    # Both are valid conditions that sep is being used a header
-                    len(stripped_line) == len(sep) or stripped_line[len(sep)] == " "
-                ):
-                    # Ensure we are tracking the header as metadata
-                    if name is not None:
-                        # Get the current header level
-                        current_header_level = sep.count("#")
-
-                        # Pop out headers of lower or same level from the stack
-                        while header_stack and header_stack[-1]["level"] >= current_header_level:
-                            # We have encountered a new header
-                            # at the same or higher level
-                            popped_header = header_stack.pop()
-                            # Clear the metadata for the
-                            # popped header in initial_metadata
-                            if popped_header["name"] in initial_metadata:
-                                initial_metadata.pop(popped_header["name"])
-
-                        # Push the current header to the stack
-                        header: HeaderType = {
-                            "level": current_header_level,
-                            "name": name,
-                            "data": stripped_line[len(sep) :].strip(),
-                        }
-                        header_stack.append(header)
-                        # Update initial_metadata with the current header
-                        initial_metadata[name] = header["data"]
-
-                    # Add the previous line to the lines_with_metadata
-                    # only if current_content is not empty
-                    if current_content:
-                        lines_with_metadata.append(
-                            {
-                                "content": "\n".join(current_content),
-                                "metadata": current_metadata.copy(),
-                            }
-                        )
-                        current_content.clear()
-
-                    break
-            else:
-                if stripped_line:
-                    current_content.append(stripped_line)
-                elif current_content:
-                    lines_with_metadata.append(
-                        {
-                            "content": "\n".join(current_content),
-                            "metadata": current_metadata.copy(),
-                        }
-                    )
-                    current_content.clear()
-
-            current_metadata = initial_metadata.copy()
-
-        if current_content:
-            lines_with_metadata.append({"content": "\n".join(current_content), "metadata": current_metadata})
-
-        # lines_with_metadata has each line with associated header metadata
-        # aggregate these into chunks based on common metadata
-        if not self.return_each_line:
-            return self.aggregate_lines_to_chunks(lines_with_metadata)
-        else:
-            return [
-                Document(page_content=chunk["content"], metadata=chunk["metadata"]) for chunk in lines_with_metadata
-            ]
-
-
-# should be in newer Python versions (3.10+)
 # @dataclass(frozen=True, kw_only=True, slots=True)
 @dataclass(frozen=True)
 class Tokenizer:

api/core/repositories/sqlalchemy_workflow_execution_repository.py

@@ -6,7 +6,6 @@ import json
 import logging
 from typing import Optional, Union
 
-from sqlalchemy import select
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 
@@ -206,44 +205,3 @@ class SQLAlchemyWorkflowExecutionRepository(WorkflowExecutionRepository):
             # Update the in-memory cache for faster subsequent lookups
             logger.debug(f"Updating cache for execution_id: {db_model.id}")
             self._execution_cache[db_model.id] = db_model
-
-    def get(self, execution_id: str) -> Optional[WorkflowExecution]:
-        """
-        Retrieve a WorkflowExecution by its ID.
-
-        First checks the in-memory cache, and if not found, queries the database.
-        If found in the database, adds it to the cache for future lookups.
-
-        Args:
-            execution_id: The workflow execution ID
-
-        Returns:
-            The WorkflowExecution instance if found, None otherwise
-        """
-        # First check the cache
-        if execution_id in self._execution_cache:
-            logger.debug(f"Cache hit for execution_id: {execution_id}")
-            # Convert cached DB model to domain model
-            cached_db_model = self._execution_cache[execution_id]
-            return self._to_domain_model(cached_db_model)
-
-        # If not in cache, query the database
-        logger.debug(f"Cache miss for execution_id: {execution_id}, querying database")
-        with self._session_factory() as session:
-            stmt = select(WorkflowRun).where(
-                WorkflowRun.id == execution_id,
-                WorkflowRun.tenant_id == self._tenant_id,
-            )
-
-            if self._app_id:
-                stmt = stmt.where(WorkflowRun.app_id == self._app_id)
-
-            db_model = session.scalar(stmt)
-            if db_model:
-                # Add DB model to cache
-                self._execution_cache[execution_id] = db_model
-
-                # Convert to domain model and return
-                return self._to_domain_model(db_model)
-
-            return None

api/core/repositories/sqlalchemy_workflow_node_execution_repository.py

@@ -7,7 +7,7 @@ import logging
 from collections.abc import Sequence
 from typing import Optional, Union
 
-from sqlalchemy import UnaryExpression, asc, delete, desc, select
+from sqlalchemy import UnaryExpression, asc, desc, select
 from sqlalchemy.engine import Engine
 from sqlalchemy.orm import sessionmaker
 
@@ -218,47 +218,6 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
                 logger.debug(f"Updating cache for node_execution_id: {db_model.node_execution_id}")
                 self._node_execution_cache[db_model.node_execution_id] = db_model
 
-    def get_by_node_execution_id(self, node_execution_id: str) -> Optional[WorkflowNodeExecution]:
-        """
-        Retrieve a NodeExecution by its node_execution_id.
-
-        First checks the in-memory cache, and if not found, queries the database.
-        If found in the database, adds it to the cache for future lookups.
-
-        Args:
-            node_execution_id: The node execution ID
-
-        Returns:
-            The NodeExecution instance if found, None otherwise
-        """
-        # First check the cache
-        if node_execution_id in self._node_execution_cache:
-            logger.debug(f"Cache hit for node_execution_id: {node_execution_id}")
-            # Convert cached DB model to domain model
-            cached_db_model = self._node_execution_cache[node_execution_id]
-            return self._to_domain_model(cached_db_model)
-
-        # If not in cache, query the database
-        logger.debug(f"Cache miss for node_execution_id: {node_execution_id}, querying database")
-        with self._session_factory() as session:
-            stmt = select(WorkflowNodeExecutionModel).where(
-                WorkflowNodeExecutionModel.node_execution_id == node_execution_id,
-                WorkflowNodeExecutionModel.tenant_id == self._tenant_id,
-            )
-
-            if self._app_id:
-                stmt = stmt.where(WorkflowNodeExecutionModel.app_id == self._app_id)
-
-            db_model = session.scalar(stmt)
-            if db_model:
-                # Add DB model to cache
-                self._node_execution_cache[node_execution_id] = db_model
-
-                # Convert to domain model and return
-                return self._to_domain_model(db_model)
-
-            return None
-
     def get_db_models_by_workflow_run(
         self,
         workflow_run_id: str,
@@ -344,68 +303,3 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
             domain_models.append(domain_model)
 
         return domain_models
-
-    def get_running_executions(self, workflow_run_id: str) -> Sequence[WorkflowNodeExecution]:
-        """
-        Retrieve all running NodeExecution instances for a specific workflow run.
-
-        This method queries the database directly and updates the cache with any
-        retrieved executions that have a node_execution_id.
-
-        Args:
-            workflow_run_id: The workflow run ID
-
-        Returns:
-            A list of running NodeExecution instances
-        """
-        with self._session_factory() as session:
-            stmt = select(WorkflowNodeExecutionModel).where(
-                WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
-                WorkflowNodeExecutionModel.tenant_id == self._tenant_id,
-                WorkflowNodeExecutionModel.status == WorkflowNodeExecutionStatus.RUNNING,
-                WorkflowNodeExecutionModel.triggered_from == WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
-            )
-
-            if self._app_id:
-                stmt = stmt.where(WorkflowNodeExecutionModel.app_id == self._app_id)
-
-            db_models = session.scalars(stmt).all()
-            domain_models = []
-
-            for model in db_models:
-                # Update cache if node_execution_id is present
-                if model.node_execution_id:
-                    self._node_execution_cache[model.node_execution_id] = model
-
-                # Convert to domain model
-                domain_model = self._to_domain_model(model)
-                domain_models.append(domain_model)
-
-            return domain_models
-
-    def clear(self) -> None:
-        """
-        Clear all WorkflowNodeExecution records for the current tenant_id and app_id.
-
-        This method deletes all WorkflowNodeExecution records that match the tenant_id
-        and app_id (if provided) associated with this repository instance.
-        It also clears the in-memory cache.
-        """
-        with self._session_factory() as session:
-            stmt = delete(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.tenant_id == self._tenant_id)
-
-            if self._app_id:
-                stmt = stmt.where(WorkflowNodeExecutionModel.app_id == self._app_id)
-
-            result = session.execute(stmt)
-            session.commit()
-
-            deleted_count = result.rowcount
-            logger.info(
-                f"Cleared {deleted_count} workflow node execution records for tenant {self._tenant_id}"
-                + (f" and app {self._app_id}" if self._app_id else "")
-            )
-
-            # Clear the in-memory cache
-            self._node_execution_cache.clear()
-            logger.info("Cleared in-memory node execution cache")

api/core/tools/__base/tool_runtime.py

@@ -4,7 +4,7 @@ from openai import BaseModel
 from pydantic import Field
 
 from core.app.entities.app_invoke_entities import InvokeFrom
-from core.tools.entities.tool_entities import ToolInvokeFrom
+from core.tools.entities.tool_entities import CredentialType, ToolInvokeFrom
 
 
 class ToolRuntime(BaseModel):
@@ -17,6 +17,7 @@ class ToolRuntime(BaseModel):
     invoke_from: Optional[InvokeFrom] = None
     tool_invoke_from: Optional[ToolInvokeFrom] = None
     credentials: dict[str, Any] = Field(default_factory=dict)
+    credential_type: CredentialType = Field(default=CredentialType.API_KEY)
     runtime_parameters: dict[str, Any] = Field(default_factory=dict)
 
 

api/core/tools/builtin_tool/provider.py

@@ -7,7 +7,13 @@ from core.helper.module_import_helper import load_single_subclass_from_source
 from core.tools.__base.tool_provider import ToolProviderController
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.builtin_tool.tool import BuiltinTool
-from core.tools.entities.tool_entities import ToolEntity, ToolProviderEntity, ToolProviderType
+from core.tools.entities.tool_entities import (
+    CredentialType,
+    OAuthSchema,
+    ToolEntity,
+    ToolProviderEntity,
+    ToolProviderType,
+)
 from core.tools.entities.values import ToolLabelEnum, default_tool_label_dict
 from core.tools.errors import (
     ToolProviderNotFoundError,
@@ -39,10 +45,18 @@ class BuiltinToolProviderController(ToolProviderController):
             credential_dict = provider_yaml.get("credentials_for_provider", {}).get(credential, {})
             credentials_schema.append(credential_dict)
 
+        oauth_schema = None
+        if provider_yaml.get("oauth_schema", None) is not None:
+            oauth_schema = OAuthSchema(
+                client_schema=provider_yaml.get("oauth_schema", {}).get("client_schema", []),
+                credentials_schema=provider_yaml.get("oauth_schema", {}).get("credentials_schema", []),
+            )
+
         super().__init__(
             entity=ToolProviderEntity(
                 identity=provider_yaml["identity"],
                 credentials_schema=credentials_schema,
+                oauth_schema=oauth_schema,
             ),
         )
 
@@ -97,10 +111,39 @@ class BuiltinToolProviderController(ToolProviderController):
 
         :return: the credentials schema
         """
-        if not self.entity.credentials_schema:
-            return []
+        return self.get_credentials_schema_by_type(CredentialType.API_KEY.value)
 
-        return self.entity.credentials_schema.copy()
+    def get_credentials_schema_by_type(self, credential_type: str) -> list[ProviderConfig]:
+        """
+        returns the credentials schema of the provider
+
+        :param credential_type: the type of the credential
+        :return: the credentials schema of the provider
+        """
+        if credential_type == CredentialType.OAUTH2.value:
+            return self.entity.oauth_schema.credentials_schema.copy() if self.entity.oauth_schema else []
+        if credential_type == CredentialType.API_KEY.value:
+            return self.entity.credentials_schema.copy() if self.entity.credentials_schema else []
+        raise ValueError(f"Invalid credential type: {credential_type}")
+
+    def get_oauth_client_schema(self) -> list[ProviderConfig]:
+        """
+        returns the oauth client schema of the provider
+
+        :return: the oauth client schema
+        """
+        return self.entity.oauth_schema.client_schema.copy() if self.entity.oauth_schema else []
+
+    def get_supported_credential_types(self) -> list[str]:
+        """
+        returns the credential support type of the provider
+        """
+        types = []
+        if self.entity.credentials_schema is not None and len(self.entity.credentials_schema) > 0:
+            types.append(CredentialType.API_KEY.value)
+        if self.entity.oauth_schema is not None and len(self.entity.oauth_schema.credentials_schema) > 0:
+            types.append(CredentialType.OAUTH2.value)
+        return types
 
     def get_tools(self) -> list[BuiltinTool]:
         """
@@ -123,7 +166,11 @@ class BuiltinToolProviderController(ToolProviderController):
 
         :return: whether the provider needs credentials
         """
-        return self.entity.credentials_schema is not None and len(self.entity.credentials_schema) != 0
+        return (
+            self.entity.credentials_schema is not None
+            and len(self.entity.credentials_schema) != 0
+            or (self.entity.oauth_schema is not None and len(self.entity.oauth_schema.credentials_schema) != 0)
+        )
 
     @property
     def provider_type(self) -> ToolProviderType:

api/core/tools/entities/api_entities.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field, field_validator
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.tools.__base.tool import ToolParameter
 from core.tools.entities.common_entities import I18nObject
-from core.tools.entities.tool_entities import ToolProviderType
+from core.tools.entities.tool_entities import CredentialType, ToolProviderType
 
 
 class ToolApiEntity(BaseModel):
@@ -87,3 +87,22 @@ class ToolProviderApiEntity(BaseModel):
     def optional_field(self, key: str, value: Any) -> dict:
         """Return dict with key-value if value is truthy, empty dict otherwise."""
         return {key: value} if value else {}
+
+
+class ToolProviderCredentialApiEntity(BaseModel):
+    id: str = Field(description="The unique id of the credential")
+    name: str = Field(description="The name of the credential")
+    provider: str = Field(description="The provider of the credential")
+    credential_type: CredentialType = Field(description="The type of the credential")
+    is_default: bool = Field(
+        default=False, description="Whether the credential is the default credential for the provider in the workspace"
+    )
+    credentials: dict = Field(description="The credentials of the provider")
+
+
+class ToolProviderCredentialInfoApiEntity(BaseModel):
+    supported_credential_types: list[str] = Field(description="The supported credential types of the provider")
+    is_oauth_custom_client_enabled: bool = Field(
+        default=False, description="Whether the OAuth custom client is enabled for the provider"
+    )
+    credentials: list[ToolProviderCredentialApiEntity] = Field(description="The credentials of the provider")

api/core/tools/entities/tool_entities.py

@@ -16,6 +16,7 @@ from core.plugin.entities.parameters import (
     cast_parameter_value,
     init_frontend_parameter,
 )
+from core.rag.entities.citation_metadata import RetrievalSourceMetadata
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.constants import TOOL_SELECTOR_MODEL_IDENTITY
 
@@ -179,6 +180,10 @@ class ToolInvokeMessage(BaseModel):
         data: Mapping[str, Any] = Field(..., description="Detailed log data")
         metadata: Optional[Mapping[str, Any]] = Field(default=None, description="The metadata of the log")
 
+    class RetrieverResourceMessage(BaseModel):
+        retriever_resources: list[RetrievalSourceMetadata] = Field(..., description="retriever resources")
+        context: str = Field(..., description="context")
+
     class MessageType(Enum):
         TEXT = "text"
         IMAGE = "image"
@@ -191,13 +196,22 @@ class ToolInvokeMessage(BaseModel):
         FILE = "file"
         LOG = "log"
         BLOB_CHUNK = "blob_chunk"
+        RETRIEVER_RESOURCES = "retriever_resources"
 
     type: MessageType = MessageType.TEXT
     """
         plain text, image url or link url
     """
     message: (
-        JsonMessage | TextMessage | BlobChunkMessage | BlobMessage | LogMessage | FileMessage | None | VariableMessage
+        JsonMessage
+        | TextMessage
+        | BlobChunkMessage
+        | BlobMessage
+        | LogMessage
+        | FileMessage
+        | None
+        | VariableMessage
+        | RetrieverResourceMessage
     )
     meta: dict[str, Any] | None = None
 
@@ -243,6 +257,7 @@ class ToolParameter(PluginParameter):
         FILES = PluginParameterType.FILES.value
         APP_SELECTOR = PluginParameterType.APP_SELECTOR.value
         MODEL_SELECTOR = PluginParameterType.MODEL_SELECTOR.value
+        ANY = PluginParameterType.ANY.value
         DYNAMIC_SELECT = PluginParameterType.DYNAMIC_SELECT.value
 
         # MCP object and array type parameters
@@ -355,10 +370,18 @@ class ToolEntity(BaseModel):
         return v or []
 
 
+class OAuthSchema(BaseModel):
+    client_schema: list[ProviderConfig] = Field(default_factory=list, description="The schema of the OAuth client")
+    credentials_schema: list[ProviderConfig] = Field(
+        default_factory=list, description="The schema of the OAuth credentials"
+    )
+
+
 class ToolProviderEntity(BaseModel):
     identity: ToolProviderIdentity
     plugin_id: Optional[str] = None
     credentials_schema: list[ProviderConfig] = Field(default_factory=list)
+    oauth_schema: Optional[OAuthSchema] = None
 
 
 class ToolProviderEntityWithPlugin(ToolProviderEntity):
@@ -438,6 +461,7 @@ class ToolSelector(BaseModel):
         options: Optional[list[PluginParameterOption]] = None
 
     provider_id: str = Field(..., description="The id of the provider")
+    credential_id: Optional[str] = Field(default=None, description="The id of the credential")
     tool_name: str = Field(..., description="The name of the tool")
     tool_description: str = Field(..., description="The description of the tool")
     tool_configuration: Mapping[str, Any] = Field(..., description="Configuration, type form")
@@ -445,3 +469,36 @@ class ToolSelector(BaseModel):
 
     def to_plugin_parameter(self) -> dict[str, Any]:
         return self.model_dump()
+
+
+class CredentialType(enum.StrEnum):
+    API_KEY = "api-key"
+    OAUTH2 = "oauth2"
+
+    def get_name(self):
+        if self == CredentialType.API_KEY:
+            return "API KEY"
+        elif self == CredentialType.OAUTH2:
+            return "AUTH"
+        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 == "api-key":
+            return cls.API_KEY
+        elif type_name == "oauth2":
+            return cls.OAUTH2
+        else:
+            raise ValueError(f"Invalid credential type: {credential_type}")

api/core/tools/plugin_tool/tool.py

@@ -44,6 +44,7 @@ class PluginTool(Tool):
             tool_provider=self.entity.identity.provider,
             tool_name=self.entity.identity.name,
             credentials=self.runtime.credentials,
+            credential_type=self.runtime.credential_type,
             tool_parameters=tool_parameters,
             conversation_id=conversation_id,
             app_id=app_id,

api/core/tools/tool_manager.py

@@ -1,15 +1,19 @@
 import json
 import logging
 import mimetypes
-from collections.abc import Generator
+import time
+from collections.abc import Generator, Mapping
 from os import listdir, path
 from threading import Lock
 from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast
 
+from pydantic import TypeAdapter
 from yarl import URL
 
 import contexts
+from core.helper.provider_cache import ToolProviderCredentialsCache
 from core.plugin.entities.plugin import ToolProviderID
+from core.plugin.impl.oauth import OAuthHandler
 from core.plugin.impl.tool import PluginToolManager
 from core.tools.__base.tool_provider import ToolProviderController
 from core.tools.__base.tool_runtime import ToolRuntime
@@ -17,14 +21,14 @@ from core.tools.mcp_tool.provider import MCPToolProviderController
 from core.tools.mcp_tool.tool import MCPTool
 from core.tools.plugin_tool.provider import PluginToolProviderController
 from core.tools.plugin_tool.tool import PluginTool
+from core.tools.utils.uuid_utils import is_valid_uuid
 from core.tools.workflow_as_tool.provider import WorkflowToolProviderController
 from core.workflow.entities.variable_pool import VariablePool
-from services.tools.mcp_tools_mange_service import MCPToolManageService
+from services.tools.mcp_tools_manage_service import MCPToolManageService
 
 if TYPE_CHECKING:
     from core.workflow.nodes.tool.entities import ToolEntity
 
-
 from configs import dify_config
 from core.agent.entities import AgentToolEntity
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -41,16 +45,17 @@ from core.tools.entities.api_entities import ToolProviderApiEntity, ToolProvider
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import (
     ApiProviderAuthType,
+    CredentialType,
     ToolInvokeFrom,
     ToolParameter,
     ToolProviderType,
 )
-from core.tools.errors import ToolNotFoundError, ToolProviderNotFoundError
+from core.tools.errors import ToolProviderNotFoundError
 from core.tools.tool_label_manager import ToolLabelManager
 from core.tools.utils.configuration import (
-    ProviderConfigEncrypter,
     ToolParameterConfigurationManager,
 )
+from core.tools.utils.encryption import create_provider_encrypter, create_tool_provider_encrypter
 from core.tools.workflow_as_tool.tool import WorkflowTool
 from extensions.ext_database import db
 from models.tools import ApiToolProvider, BuiltinToolProvider, MCPToolProvider, WorkflowToolProvider
@@ -68,8 +73,11 @@ class ToolManager:
     @classmethod
     def get_hardcoded_provider(cls, provider: str) -> BuiltinToolProviderController:
         """
+
         get the hardcoded provider
+
         """
+
         if len(cls._hardcoded_providers) == 0:
             # init the builtin providers
             cls.load_hardcoded_providers_cache()
@@ -113,7 +121,12 @@ class ToolManager:
             contexts.plugin_tool_providers.set({})
             contexts.plugin_tool_providers_lock.set(Lock())
 
+        plugin_tool_providers = contexts.plugin_tool_providers.get()
+        if provider in plugin_tool_providers:
+            return plugin_tool_providers[provider]
+
         with contexts.plugin_tool_providers_lock.get():
+            # double check
             plugin_tool_providers = contexts.plugin_tool_providers.get()
             if provider in plugin_tool_providers:
                 return plugin_tool_providers[provider]
@@ -131,25 +144,7 @@ class ToolManager:
             )
 
             plugin_tool_providers[provider] = controller
-
-        return controller
-
-    @classmethod
-    def get_builtin_tool(cls, provider: str, tool_name: str, tenant_id: str) -> BuiltinTool | PluginTool | None:
-        """
-        get the builtin tool
-
-        :param provider: the name of the provider
-        :param tool_name: the name of the tool
-        :param tenant_id: the id of the tenant
-        :return: the provider, the tool
-        """
-        provider_controller = cls.get_builtin_provider(provider, tenant_id)
-        tool = provider_controller.get_tool(tool_name)
-        if tool is None:
-            raise ToolNotFoundError(f"tool {tool_name} not found")
-
-        return tool
+            return controller
 
     @classmethod
     def get_tool_runtime(
@@ -160,6 +155,7 @@ class ToolManager:
         tenant_id: str,
         invoke_from: InvokeFrom = InvokeFrom.DEBUGGER,
         tool_invoke_from: ToolInvokeFrom = ToolInvokeFrom.AGENT,
+        credential_id: Optional[str] = None,
     ) -> Union[BuiltinTool, PluginTool, ApiTool, WorkflowTool, MCPTool]:
         """
         get the tool runtime
@@ -170,6 +166,7 @@ class ToolManager:
         :param tenant_id: the tenant id
         :param invoke_from: invoke from
         :param tool_invoke_from: the tool invoke from
+        :param credential_id: the credential id
 
         :return: the tool
         """
@@ -193,49 +190,105 @@ class ToolManager:
                         )
                     ),
                 )
-
+            builtin_provider = None
             if isinstance(provider_controller, PluginToolProviderController):
                 provider_id_entity = ToolProviderID(provider_id)
-                # get credentials
-                builtin_provider: BuiltinToolProvider | None = (
-                    db.session.query(BuiltinToolProvider)
-                    .filter(
-                        BuiltinToolProvider.tenant_id == tenant_id,
-                        (BuiltinToolProvider.provider == str(provider_id_entity))
-                        | (BuiltinToolProvider.provider == provider_id_entity.provider_name),
-                    )
-                    .first()
-                )
+                # get specific credentials
+                if is_valid_uuid(credential_id):
+                    try:
+                        builtin_provider = (
+                            db.session.query(BuiltinToolProvider)
+                            .filter(
+                                BuiltinToolProvider.tenant_id == tenant_id,
+                                BuiltinToolProvider.id == credential_id,
+                            )
+                            .first()
+                        )
+                    except Exception as e:
+                        builtin_provider = None
+                        logger.info(f"Error getting builtin provider {credential_id}:{e}", exc_info=True)
+                    # if the provider has been deleted, raise an error
+                    if builtin_provider is None:
+                        raise ToolProviderNotFoundError(f"provider has been deleted: {credential_id}")
 
+                # fallback to the default provider
                 if builtin_provider is None:
-                    raise ToolProviderNotFoundError(f"builtin provider {provider_id} not found")
+                    # use the default provider
+                    builtin_provider = (
+                        db.session.query(BuiltinToolProvider)
+                        .filter(
+                            BuiltinToolProvider.tenant_id == tenant_id,
+                            (BuiltinToolProvider.provider == str(provider_id_entity))
+                            | (BuiltinToolProvider.provider == provider_id_entity.provider_name),
+                        )
+                        .order_by(BuiltinToolProvider.is_default.desc(), BuiltinToolProvider.created_at.asc())
+                        .first()
+                    )
+                    if builtin_provider is None:
+                        raise ToolProviderNotFoundError(f"no default provider for {provider_id}")
             else:
                 builtin_provider = (
                     db.session.query(BuiltinToolProvider)
                     .filter(BuiltinToolProvider.tenant_id == tenant_id, (BuiltinToolProvider.provider == provider_id))
+                    .order_by(BuiltinToolProvider.is_default.desc(), BuiltinToolProvider.created_at.asc())
                     .first()
                 )
 
                 if builtin_provider is None:
                     raise ToolProviderNotFoundError(f"builtin provider {provider_id} not found")
 
-            # decrypt the credentials
-            credentials = builtin_provider.credentials
-            tool_configuration = ProviderConfigEncrypter(
+            encrypter, _ = create_provider_encrypter(
                 tenant_id=tenant_id,
-                config=[x.to_basic_provider_config() for x in provider_controller.get_credentials_schema()],
-                provider_type=provider_controller.provider_type.value,
-                provider_identity=provider_controller.entity.identity.name,
+                config=[
+                    x.to_basic_provider_config()
+                    for x in provider_controller.get_credentials_schema_by_type(builtin_provider.credential_type)
+                ],
+                cache=ToolProviderCredentialsCache(
+                    tenant_id=tenant_id, provider=provider_id, credential_id=builtin_provider.id
+                ),
             )
 
-            decrypted_credentials = tool_configuration.decrypt(credentials)
+            # decrypt the credentials
+            decrypted_credentials: Mapping[str, Any] = encrypter.decrypt(builtin_provider.credentials)
+
+            # check if the credentials is expired
+            if builtin_provider.expires_at != -1 and (builtin_provider.expires_at - 60) < int(time.time()):
+                # TODO: circular import
+                from services.tools.builtin_tools_manage_service import BuiltinToolManageService
+
+                # refresh the credentials
+                tool_provider = ToolProviderID(provider_id)
+                provider_name = tool_provider.provider_name
+                redirect_uri = f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{provider_id}/tool/callback"
+                system_credentials = BuiltinToolManageService.get_oauth_client(tenant_id, provider_id)
+                oauth_handler = OAuthHandler()
+                # refresh the credentials
+                refreshed_credentials = oauth_handler.refresh_credentials(
+                    tenant_id=tenant_id,
+                    user_id=builtin_provider.user_id,
+                    plugin_id=tool_provider.plugin_id,
+                    provider=provider_name,
+                    redirect_uri=redirect_uri,
+                    system_credentials=system_credentials or {},
+                    credentials=decrypted_credentials,
+                )
+                # update the credentials
+                builtin_provider.encrypted_credentials = (
+                    TypeAdapter(dict[str, Any])
+                    .dump_json(encrypter.encrypt(dict(refreshed_credentials.credentials)))
+                    .decode("utf-8")
+                )
+                builtin_provider.expires_at = refreshed_credentials.expires_at
+                db.session.commit()
+                decrypted_credentials = refreshed_credentials.credentials
 
             return cast(
                 BuiltinTool,
                 builtin_tool.fork_tool_runtime(
                     runtime=ToolRuntime(
                         tenant_id=tenant_id,
-                        credentials=decrypted_credentials,
+                        credentials=dict(decrypted_credentials),
+                        credential_type=CredentialType.of(builtin_provider.credential_type),
                         runtime_parameters={},
                         invoke_from=invoke_from,
                         tool_invoke_from=tool_invoke_from,
@@ -245,22 +298,16 @@ class ToolManager:
 
         elif provider_type == ToolProviderType.API:
             api_provider, credentials = cls.get_api_provider_controller(tenant_id, provider_id)
-
-            # decrypt the credentials
-            tool_configuration = ProviderConfigEncrypter(
+            encrypter, _ = create_tool_provider_encrypter(
                 tenant_id=tenant_id,
-                config=[x.to_basic_provider_config() for x in api_provider.get_credentials_schema()],
-                provider_type=api_provider.provider_type.value,
-                provider_identity=api_provider.entity.identity.name,
+                controller=api_provider,
             )
-            decrypted_credentials = tool_configuration.decrypt(credentials)
-
             return cast(
                 ApiTool,
                 api_provider.get_tool(tool_name).fork_tool_runtime(
                     runtime=ToolRuntime(
                         tenant_id=tenant_id,
-                        credentials=decrypted_credentials,
+                        credentials=encrypter.decrypt(credentials),
                         invoke_from=invoke_from,
                         tool_invoke_from=tool_invoke_from,
                     )
@@ -320,6 +367,7 @@ class ToolManager:
             tenant_id=tenant_id,
             invoke_from=invoke_from,
             tool_invoke_from=ToolInvokeFrom.AGENT,
+            credential_id=agent_tool.credential_id,
         )
         runtime_parameters = {}
         parameters = tool_entity.get_merged_runtime_parameters()
@@ -362,6 +410,7 @@ class ToolManager:
             tenant_id=tenant_id,
             invoke_from=invoke_from,
             tool_invoke_from=ToolInvokeFrom.WORKFLOW,
+            credential_id=workflow_tool.credential_id,
         )
 
         parameters = tool_runtime.get_merged_runtime_parameters()
@@ -391,6 +440,7 @@ class ToolManager:
         provider: str,
         tool_name: str,
         tool_parameters: dict[str, Any],
+        credential_id: Optional[str] = None,
     ) -> Tool:
         """
         get tool runtime from plugin
@@ -402,6 +452,7 @@ class ToolManager:
             tenant_id=tenant_id,
             invoke_from=InvokeFrom.SERVICE_API,
             tool_invoke_from=ToolInvokeFrom.PLUGIN,
+            credential_id=credential_id,
         )
         runtime_parameters = {}
         parameters = tool_entity.get_merged_runtime_parameters()
@@ -551,6 +602,22 @@ class ToolManager:
 
         return cls._builtin_tools_labels[tool_name]
 
+    @classmethod
+    def list_default_builtin_providers(cls, tenant_id: str) -> list[BuiltinToolProvider]:
+        """
+        list all the builtin providers
+        """
+        # according to multi credentials, select the one with is_default=True first, then created_at oldest
+        # for compatibility with old version
+        sql = """
+                SELECT DISTINCT ON (tenant_id, provider) id
+                FROM tool_builtin_providers
+                WHERE tenant_id = :tenant_id
+                ORDER BY tenant_id, provider, is_default DESC, created_at DESC
+                """
+        ids = [row.id for row in db.session.execute(db.text(sql), {"tenant_id": tenant_id}).all()]
+        return db.session.query(BuiltinToolProvider).filter(BuiltinToolProvider.id.in_(ids)).all()
+
     @classmethod
     def list_providers_from_api(
         cls, user_id: str, tenant_id: str, typ: ToolProviderTypeApiLiteral
@@ -565,21 +632,13 @@ class ToolManager:
 
         with db.session.no_autoflush:
             if "builtin" in filters:
-                # get builtin providers
                 builtin_providers = cls.list_builtin_providers(tenant_id)
 
-                # get db builtin providers
-                db_builtin_providers: list[BuiltinToolProvider] = (
-                    db.session.query(BuiltinToolProvider).filter(BuiltinToolProvider.tenant_id == tenant_id).all()
-                )
-
-                # rewrite db_builtin_providers
-                for db_provider in db_builtin_providers:
-                    tool_provider_id = str(ToolProviderID(db_provider.provider))
-                    db_provider.provider = tool_provider_id
-
-                def find_db_builtin_provider(provider):
-                    return next((x for x in db_builtin_providers if x.provider == provider), None)
+                # key: provider name, value: provider
+                db_builtin_providers = {
+                    str(ToolProviderID(provider.provider)): provider
+                    for provider in cls.list_default_builtin_providers(tenant_id)
+                }
 
                 # append builtin providers
                 for provider in builtin_providers:
@@ -591,10 +650,9 @@ class ToolManager:
                         name_func=lambda x: x.identity.name,
                     ):
                         continue
-
                     user_provider = ToolTransformService.builtin_provider_to_user_provider(
                         provider_controller=provider,
-                        db_provider=find_db_builtin_provider(provider.entity.identity.name),
+                        db_provider=db_builtin_providers.get(provider.entity.identity.name),
                         decrypt_credentials=False,
                     )
 
@@ -604,7 +662,6 @@ class ToolManager:
                         result_providers[f"builtin_provider.{user_provider.name}"] = user_provider
 
             # get db api providers
-
             if "api" in filters:
                 db_api_providers: list[ApiToolProvider] = (
                     db.session.query(ApiToolProvider).filter(ApiToolProvider.tenant_id == tenant_id).all()
@@ -764,15 +821,12 @@ class ToolManager:
             auth_type,
         )
         # init tool configuration
-        tool_configuration = ProviderConfigEncrypter(
+        encrypter, _ = create_tool_provider_encrypter(
             tenant_id=tenant_id,
-            config=[x.to_basic_provider_config() for x in controller.get_credentials_schema()],
-            provider_type=controller.provider_type.value,
-            provider_identity=controller.entity.identity.name,
+            controller=controller,
         )
 
-        decrypted_credentials = tool_configuration.decrypt(credentials)
-        masked_credentials = tool_configuration.mask_tool_credentials(decrypted_credentials)
+        masked_credentials = encrypter.mask_tool_credentials(encrypter.decrypt(credentials))
 
         try:
             icon = json.loads(provider_obj.icon)

api/core/tools/utils/configuration.py

@@ -1,12 +1,8 @@
 from copy import deepcopy
 from typing import Any
 
-from pydantic import BaseModel
-
-from core.entities.provider_entities import BasicProviderConfig
 from core.helper import encrypter
 from core.helper.tool_parameter_cache import ToolParameterCache, ToolParameterCacheType
-from core.helper.tool_provider_cache import ToolProviderCredentialsCache, ToolProviderCredentialsCacheType
 from core.tools.__base.tool import Tool
 from core.tools.entities.tool_entities import (
     ToolParameter,
@@ -14,110 +10,6 @@ from core.tools.entities.tool_entities import (
 )
 
 
-class ProviderConfigEncrypter(BaseModel):
-    tenant_id: str
-    config: list[BasicProviderConfig]
-    provider_type: str
-    provider_identity: str
-
-    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_tool_credentials(self, data: dict[str, Any]) -> dict[str, Any]:
-        """
-        mask tool 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 decrypt(self, data: dict[str, str], use_cache: bool = True) -> dict[str, str]:
-        """
-        decrypt tool credentials with tenant id
-
-        return a deep copy of credentials with decrypted values
-        """
-        if use_cache:
-            cache = ToolProviderCredentialsCache(
-                tenant_id=self.tenant_id,
-                identity_id=f"{self.provider_type}.{self.provider_identity}",
-                cache_type=ToolProviderCredentialsCacheType.PROVIDER,
-            )
-            cached_credentials = 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:
-                    try:
-                        # 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])
-                    except Exception:
-                        pass
-
-        if use_cache:
-            cache.set(data)
-        return data
-
-    def delete_tool_credentials_cache(self):
-        cache = ToolProviderCredentialsCache(
-            tenant_id=self.tenant_id,
-            identity_id=f"{self.provider_type}.{self.provider_identity}",
-            cache_type=ToolProviderCredentialsCacheType.PROVIDER,
-        )
-        cache.delete()
-
-
 class ToolParameterConfigurationManager:
     """
     Tool parameter configuration manager