Merge branch 'main' into feat/model-auth

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

.devcontainer/post_create_command.sh

@@ -5,7 +5,7 @@ cd web && pnpm install
 pipx install uv
 
 echo 'alias start-api="cd /workspaces/dify/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug"' >> ~/.bashrc
-echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion"' >> ~/.bashrc
+echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage"' >> ~/.bashrc
 echo 'alias start-web="cd /workspaces/dify/web && pnpm dev"' >> ~/.bashrc
 echo 'alias start-web-prod="cd /workspaces/dify/web && pnpm build && pnpm start"' >> ~/.bashrc
 echo 'alias start-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d"' >> ~/.bashrc

.github/ISSUE_TEMPLATE/refactor.yml

@@ -0,0 +1,44 @@
+name: "✨ Refactor"
+description: Refactor existing code for improved readability and maintainability.
+title: "[Chore/Refactor] "
+labels:
+  - refactor
+body:
+  - type: checkboxes
+    attributes:
+      label: Self Checks
+      description: "To make sure we get to you in time, please check the following :)"
+      options:
+        - label: I have read the [Contributing Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) and [Language Policy](https://github.com/langgenius/dify/issues/1542).
+          required: true
+        - label: This is only for refactoring, if you would like to ask a question, please head to [Discussions](https://github.com/langgenius/dify/discussions/categories/general).
+          required: true
+        - label: I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones.
+          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
+  - type: textarea
+    id: description
+    attributes:
+      label: Description
+      placeholder: "Describe the refactor you are proposing."
+    validations:
+      required: true
+  - type: textarea
+    id: motivation
+    attributes:
+      label: Motivation
+      placeholder: "Explain why this refactor is necessary."
+    validations:
+      required: false
+  - type: textarea
+    id: additional-context
+    attributes:
+      label: Additional Context
+      placeholder: "Add any other context or screenshots about the request here."
+    validations:
+      required: false

.github/actions/setup-uv/action.yml

@@ -8,7 +8,7 @@ inputs:
   uv-version:
     description: UV version to set up
     required: true
-    default: '~=0.7.11'
+    default: '0.8.9'
   uv-lockfile:
     description: Path to the UV lockfile to restore cache from
     required: true
@@ -26,7 +26,7 @@ runs:
         python-version: ${{ inputs.python-version }}
 
     - name: Install uv
-      uses: astral-sh/setup-uv@v5
+      uses: astral-sh/setup-uv@v6
       with:
         version: ${{ inputs.uv-version }}
         python-version: ${{ inputs.python-version }}

.github/workflows/api-tests.yml

@@ -99,3 +99,6 @@ jobs:
 
       - name: Run Tool
         run: uv run --project api bash dev/pytest/pytest_tools.sh
+
+      - name: Run TestContainers
+        run: uv run --project api bash dev/pytest/pytest_testcontainers.sh

.github/workflows/autofix.yml

@@ -9,6 +9,7 @@ permissions:
 
 jobs:
   autofix:
+    if: github.repository == 'langgenius/dify'
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

.github/workflows/build-push.yml

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

.github/workflows/style.yml

@@ -82,7 +82,7 @@ jobs:
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
-          version: 10
+          package_json_file: web/package.json
           run_install: false
 
       - name: Setup NodeJS
@@ -95,10 +95,12 @@ jobs:
 
       - name: Web dependencies
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm run lint
 
   docker-compose-template:

.github/workflows/translate-i18n-base-on-english.yml

@@ -1,13 +1,18 @@
 name: Check i18n Files and Create PR
 
 on:
-  pull_request:
-    types: [closed]
+  push:
     branches: [main]
+    paths:
+      - 'web/i18n/en-US/*.ts'
+
+permissions:
+  contents: write
+  pull-requests: write
 
 jobs:
   check-and-update:
-    if: github.event.pull_request.merged == true
+    if: github.repository == 'langgenius/dify'
     runs-on: ubuntu-latest
     defaults:
       run:
@@ -15,8 +20,8 @@ jobs:
     steps:
       - uses: actions/checkout@v4
         with:
-          fetch-depth: 2 # last 2 commits
-          persist-credentials: false
+          fetch-depth: 2
+          token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Check for file changes in i18n/en-US
         id: check_files
@@ -27,6 +32,13 @@ jobs:
           echo "Changed files: $changed_files"
           if [ -n "$changed_files" ]; then
             echo "FILES_CHANGED=true" >> $GITHUB_ENV
+            file_args=""
+            for file in $changed_files; do
+              filename=$(basename "$file" .ts)
+              file_args="$file_args --file=$filename"
+            done
+            echo "FILE_ARGS=$file_args" >> $GITHUB_ENV
+            echo "File arguments: $file_args"
           else
             echo "FILES_CHANGED=false" >> $GITHUB_ENV
           fi
@@ -34,7 +46,7 @@ jobs:
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
-          version: 10
+          package_json_file: web/package.json
           run_install: false
 
       - name: Set up Node.js
@@ -47,16 +59,19 @@ jobs:
 
       - name: Install dependencies
         if: env.FILES_CHANGED == 'true'
+        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
-      - name: Run npm script
+      - name: Generate i18n translations
         if: env.FILES_CHANGED == 'true'
-        run: pnpm run auto-gen-i18n
+        working-directory: ./web
+        run: pnpm run auto-gen-i18n ${{ env.FILE_ARGS }}
 
       - name: Create Pull Request
         if: env.FILES_CHANGED == 'true'
         uses: peter-evans/create-pull-request@v6
         with:
+          token: ${{ secrets.GITHUB_TOKEN }}
           commit-message: Update i18n files based on en-US changes
           title: 'chore: translate i18n files'
           body: This PR was automatically created to update i18n files based on changes in en-US locale.

.github/workflows/web-tests.yml

@@ -35,7 +35,7 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         uses: pnpm/action-setup@v4
         with:
-          version: 10
+          package_json_file: web/package.json
           run_install: false
 
       - name: Setup Node.js
@@ -48,8 +48,10 @@ jobs:
 
       - name: Install dependencies
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm install --frozen-lockfile
 
       - name: Run tests
         if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
         run: pnpm test

.gitignore

@@ -197,6 +197,8 @@ sdks/python-client/dify_client.egg-info
 !.vscode/README.md
 pyrightconfig.json
 api/.vscode
+# vscode Code History Extension
+.history
 
 .idea/
 
@@ -215,3 +217,4 @@ mise.toml
 # AI Assistant
 .roo/
 api/.env.backup
+/clickzetta

CLAUDE.md

@@ -0,0 +1,83 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Project Overview
+
+Dify is an open-source platform for developing LLM applications with an intuitive interface combining agentic AI workflows, RAG pipelines, agent capabilities, and model management.
+
+The codebase consists of:
+- **Backend API** (`/api`): Python Flask application with Domain-Driven Design architecture
+- **Frontend Web** (`/web`): Next.js 15 application with TypeScript and React 19
+- **Docker deployment** (`/docker`): Containerized deployment configurations
+
+## Development Commands
+
+### Backend (API)
+
+All Python commands must be prefixed with `uv run --project api`:
+
+```bash
+# Start development servers
+./dev/start-api                   # Start API server
+./dev/start-worker                # Start Celery worker
+
+# Run tests
+uv run --project api pytest      # Run all tests
+uv run --project api pytest tests/unit_tests/     # Unit tests only
+uv run --project api pytest tests/integration_tests/  # Integration tests
+
+# Code quality
+./dev/reformat                    # Run all formatters and linters
+uv run --project api ruff check --fix ./    # Fix linting issues
+uv run --project api ruff format ./         # Format code
+uv run --project api mypy .                 # Type checking
+```
+
+### Frontend (Web)
+
+```bash
+cd web
+pnpm lint                         # Run ESLint
+pnpm eslint-fix                   # Fix ESLint issues
+pnpm test                         # Run Jest tests
+```
+
+## Testing Guidelines
+
+### Backend Testing
+- Use `pytest` for all backend tests
+- Write tests first (TDD approach)
+- Test structure: Arrange-Act-Assert
+
+## Code Style Requirements
+
+### Python
+- Use type hints for all functions and class attributes
+- No `Any` types unless absolutely necessary
+- Implement special methods (`__repr__`, `__str__`) appropriately
+
+### TypeScript/JavaScript  
+- Strict TypeScript configuration
+- ESLint with Prettier integration
+- Avoid `any` type
+
+## Important Notes
+
+- **Environment Variables**: Always use UV for Python commands: `uv run --project api <command>`
+- **Comments**: Only write meaningful comments that explain "why", not "what"
+- **File Creation**: Always prefer editing existing files over creating new ones
+- **Documentation**: Don't create documentation files unless explicitly requested
+- **Code Quality**: Always run `./dev/reformat` before committing backend changes
+
+## Common Development Tasks
+
+### Adding a New API Endpoint
+1. Create controller in `/api/controllers/`
+2. Add service logic in `/api/services/`
+3. Update routes in controller's `__init__.py`
+4. Write tests in `/api/tests/`
+
+## Project-Specific Conventions
+
+- All async tasks use Celery with Redis as broker

README.md

@@ -225,7 +225,8 @@ Deploy Dify to AWS with [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS
 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Using Alibaba Cloud Computing Nest
 

README_AR.md

@@ -208,7 +208,8 @@ docker compose up -d
 
 ##### AWS
 
-- [AWS CDK بواسطة @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK بواسطة @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK بواسطة @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### استخدام Alibaba Cloud للنشر
  [بسرعة نشر Dify إلى سحابة علي بابا مع عش الحوسبة السحابية علي بابا](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

README_BN.md

@@ -225,7 +225,8 @@ GitHub-এ ডিফাইকে স্টার দিয়ে রাখুন
 
 ##### AWS
 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud ব্যবহার করে ডিপ্লয়
 

README_CN.md

@@ -223,7 +223,8 @@ docker compose up -d
 使用 [CDK](https://aws.amazon.com/cdk/) 将 Dify 部署到 AWS
 
 ##### AWS 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### 使用 阿里云计算巢 部署
 

README_DE.md

@@ -220,7 +220,8 @@ Stellen Sie Dify mit nur einem Klick mithilfe von [terraform](https://www.terraf
 Bereitstellung von Dify auf AWS mit [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud 
 

README_ES.md

@@ -220,7 +220,8 @@ Despliega Dify en una plataforma en la nube con un solo clic utilizando [terrafo
 Despliegue Dify en AWS usando [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK por @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_FR.md

@@ -218,7 +218,8 @@ Déployez Dify sur une plateforme cloud en un clic en utilisant [terraform](http
 Déployez Dify sur AWS en utilisant [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK par @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK par @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK par @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_JA.md

@@ -219,7 +219,8 @@ docker compose up -d
 [CDK](https://aws.amazon.com/cdk/) を使用して、DifyをAWSにデプロイします
 
 ##### AWS
-- [@KevinZhaoによるAWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [@KevinZhaoによるAWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [@tmokmssによるAWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

README_KL.md

@@ -218,7 +218,8 @@ wa'logh nIqHom neH ghun deployment toy'wI' [terraform](https://www.terraform.io/
 wa'logh nIqHom neH ghun deployment toy'wI' [CDK](https://aws.amazon.com/cdk/) lo'laH.
 
 ##### AWS 
-- [AWS CDK qachlot @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK qachlot @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK qachlot @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_KR.md

@@ -212,7 +212,8 @@ Dify를 Kubernetes에 배포하고 프리미엄 스케일링 설정을 구성했
 [CDK](https://aws.amazon.com/cdk/)를 사용하여 AWS에 Dify 배포
 
 ##### AWS 
-- [KevinZhao의 AWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [KevinZhao의 AWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [tmokmss의 AWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_PT.md

@@ -217,7 +217,8 @@ Implante o Dify na Plataforma Cloud com um único clique usando [terraform](http
 Implante o Dify na AWS usando [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK por @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK por @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_SI.md

@@ -218,7 +218,8 @@ namestite Dify v Cloud Platform z enim klikom z uporabo [terraform](https://www.
 Uvedite Dify v AWS z uporabo [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK by @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK by @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_TR.md

@@ -211,7 +211,8 @@ Dify'ı bulut platformuna tek tıklamayla dağıtın [terraform](https://www.ter
 [CDK](https://aws.amazon.com/cdk/) kullanarak Dify'ı AWS'ye dağıtın
 
 ##### AWS 
-- [AWS CDK tarafından @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK tarafından @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK tarafından @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### Alibaba Cloud
 

README_TW.md

@@ -223,7 +223,8 @@ Dify 的所有功能都提供相應的 API，因此您可以輕鬆地將 Dify
 
 ### AWS
 
-- [由 @KevinZhao 提供的 AWS CDK](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [由 @KevinZhao 提供的 AWS CDK (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [由 @tmokmss 提供的 AWS CDK (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 #### 使用 阿里云计算巢進行部署
 

README_VI.md

@@ -213,7 +213,8 @@ Triển khai Dify lên nền tảng đám mây với một cú nhấp chuột b
 Triển khai Dify trên AWS bằng [CDK](https://aws.amazon.com/cdk/)
 
 ##### AWS 
-- [AWS CDK bởi @KevinZhao](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK bởi @KevinZhao (EKS based)](https://github.com/aws-samples/solution-for-deploying-dify-on-aws)
+- [AWS CDK bởi @tmokmss (ECS based)](https://github.com/aws-samples/dify-self-hosted-on-aws)
 
 
 #### Alibaba Cloud

api/.env.example

@@ -42,6 +42,15 @@ REDIS_PORT=6379
 REDIS_USERNAME=
 REDIS_PASSWORD=difyai123456
 REDIS_USE_SSL=false
+# SSL configuration for Redis (when REDIS_USE_SSL=true)
+REDIS_SSL_CERT_REQS=CERT_NONE
+# Options: CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
+REDIS_SSL_CA_CERTS=
+# Path to CA certificate file for SSL verification
+REDIS_SSL_CERTFILE=
+# Path to client certificate file for SSL authentication
+REDIS_SSL_KEYFILE=
+# Path to client private key file for SSL authentication
 REDIS_DB=0
 
 # redis Sentinel configuration.

api/Dockerfile

@@ -4,7 +4,7 @@ FROM python:3.12-slim-bookworm AS base
 WORKDIR /app/api
 
 # Install uv
-ENV UV_VERSION=0.7.11
+ENV UV_VERSION=0.8.9
 
 RUN pip install --no-cache-dir uv==${UV_VERSION}
 
@@ -19,7 +19,7 @@ RUN apt-get update \
 
 # Install Python dependencies
 COPY pyproject.toml uv.lock ./
-RUN uv sync --locked
+RUN uv sync --locked --no-dev
 
 # production stage
 FROM base AS production

api/README.md

@@ -74,7 +74,7 @@
 10. If you need to handle and debug the async tasks (e.g. dataset importing and documents indexing), please start the worker service.
 
    ```bash
-   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin
+   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin,workflow_storage
    ```
 
    Addition, if you want to debug the celery scheduled tasks, you can use the following command in another terminal:

api/app_factory.py

@@ -51,6 +51,7 @@ def initialize_extensions(app: DifyApp):
         ext_login,
         ext_mail,
         ext_migrate,
+        ext_orjson,
         ext_otel,
         ext_proxy_fix,
         ext_redis,
@@ -67,6 +68,7 @@ def initialize_extensions(app: DifyApp):
         ext_logging,
         ext_warnings,
         ext_import_modules,
+        ext_orjson,
         ext_set_secretkey,
         ext_compress,
         ext_code_based_extension,

api/commands.py

@@ -5,10 +5,11 @@ import secrets
 from typing import Any, Optional
 
 import click
+import sqlalchemy as sa
 from flask import current_app
 from pydantic import TypeAdapter
 from sqlalchemy import select
-from werkzeug.exceptions import NotFound
+from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from constants.languages import languages
@@ -35,6 +36,7 @@ from services.account_service import AccountService, RegisterService, TenantServ
 from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpiredLogs
 from services.plugin.data_migration import PluginDataMigration
 from services.plugin.plugin_migration import PluginMigration
+from tasks.remove_app_and_related_data_task import delete_draft_variables_batch
 
 
 @click.command("reset-password", help="Reset the account password.")
@@ -180,8 +182,8 @@ def migrate_annotation_vector_database():
             )
             if not apps:
                 break
-        except NotFound:
-            break
+        except SQLAlchemyError:
+            raise
 
         page += 1
         for app in apps:
@@ -307,8 +309,8 @@ def migrate_knowledge_vector_database():
             )
 
             datasets = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
-        except NotFound:
-            break
+        except SQLAlchemyError:
+            raise
 
         page += 1
         for dataset in datasets:
@@ -457,7 +459,7 @@ def convert_to_agent_apps():
         """
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query))
+            rs = conn.execute(sa.text(sql_query))
 
             apps = []
             for i in rs:
@@ -560,8 +562,8 @@ def old_metadata_migration():
                 .order_by(DatasetDocument.created_at.desc())
             )
             documents = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
-        except NotFound:
-            break
+        except SQLAlchemyError:
+            raise
         if not documents:
             break
         for document in documents:
@@ -702,7 +704,7 @@ def fix_app_site_missing():
         sql = """select apps.id as id from apps left join sites on sites.app_id=apps.id
 where sites.id is null limit 1000"""
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql))
+            rs = conn.execute(sa.text(sql))
 
             processed_count = 0
             for i in rs:
@@ -916,7 +918,7 @@ def clear_orphaned_file_records(force: bool):
         )
         orphaned_message_files = []
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(query))
+            rs = conn.execute(sa.text(query))
             for i in rs:
                 orphaned_message_files.append({"id": str(i[0]), "message_id": str(i[1])})
 
@@ -937,7 +939,7 @@ def clear_orphaned_file_records(force: bool):
             click.echo(click.style("- Deleting orphaned message_files records", fg="white"))
             query = "DELETE FROM message_files WHERE id IN :ids"
             with db.engine.begin() as conn:
-                conn.execute(db.text(query), {"ids": tuple([record["id"] for record in orphaned_message_files])})
+                conn.execute(sa.text(query), {"ids": tuple([record["id"] for record in orphaned_message_files])})
             click.echo(
                 click.style(f"Removed {len(orphaned_message_files)} orphaned message_files records.", fg="green")
             )
@@ -954,7 +956,7 @@ def clear_orphaned_file_records(force: bool):
             click.echo(click.style(f"- Listing file records in table {files_table['table']}", fg="white"))
             query = f"SELECT {files_table['id_column']}, {files_table['key_column']} FROM {files_table['table']}"
             with db.engine.begin() as conn:
-                rs = conn.execute(db.text(query))
+                rs = conn.execute(sa.text(query))
             for i in rs:
                 all_files_in_tables.append({"table": files_table["table"], "id": str(i[0]), "key": i[1]})
         click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))
@@ -974,7 +976,7 @@ def clear_orphaned_file_records(force: bool):
                     f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
                 )
                 with db.engine.begin() as conn:
-                    rs = conn.execute(db.text(query))
+                    rs = conn.execute(sa.text(query))
                 for i in rs:
                     all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
             elif ids_table["type"] == "text":
@@ -989,7 +991,7 @@ def clear_orphaned_file_records(force: bool):
                     f"FROM {ids_table['table']}"
                 )
                 with db.engine.begin() as conn:
-                    rs = conn.execute(db.text(query))
+                    rs = conn.execute(sa.text(query))
                 for i in rs:
                     for j in i[0]:
                         all_ids_in_tables.append({"table": ids_table["table"], "id": j})
@@ -1008,7 +1010,7 @@ def clear_orphaned_file_records(force: bool):
                     f"FROM {ids_table['table']}"
                 )
                 with db.engine.begin() as conn:
-                    rs = conn.execute(db.text(query))
+                    rs = conn.execute(sa.text(query))
                 for i in rs:
                     for j in i[0]:
                         all_ids_in_tables.append({"table": ids_table["table"], "id": j})
@@ -1037,7 +1039,7 @@ def clear_orphaned_file_records(force: bool):
             click.echo(click.style(f"- Deleting orphaned file records in table {files_table['table']}", fg="white"))
             query = f"DELETE FROM {files_table['table']} WHERE {files_table['id_column']} IN :ids"
             with db.engine.begin() as conn:
-                conn.execute(db.text(query), {"ids": tuple(orphaned_files)})
+                conn.execute(sa.text(query), {"ids": tuple(orphaned_files)})
     except Exception as e:
         click.echo(click.style(f"Error deleting orphaned file records: {str(e)}", fg="red"))
         return
@@ -1107,7 +1109,7 @@ def remove_orphaned_files_on_storage(force: bool):
             click.echo(click.style(f"- Listing files from table {files_table['table']}", fg="white"))
             query = f"SELECT {files_table['key_column']} FROM {files_table['table']}"
             with db.engine.begin() as conn:
-                rs = conn.execute(db.text(query))
+                rs = conn.execute(sa.text(query))
             for i in rs:
                 all_files_in_tables.append(str(i[0]))
         click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))
@@ -1201,3 +1203,138 @@ def setup_system_tool_oauth_client(provider, client_params):
     db.session.add(oauth_client)
     db.session.commit()
     click.echo(click.style(f"OAuth client params setup successfully. id: {oauth_client.id}", fg="green"))
+
+
+def _find_orphaned_draft_variables(batch_size: int = 1000) -> list[str]:
+    """
+    Find draft variables that reference non-existent apps.
+
+    Args:
+        batch_size: Maximum number of orphaned app IDs to return
+
+    Returns:
+        List of app IDs that have draft variables but don't exist in the apps table
+    """
+    query = """
+        SELECT DISTINCT wdv.app_id
+        FROM workflow_draft_variables AS wdv
+        WHERE NOT EXISTS(
+            SELECT 1 FROM apps WHERE apps.id = wdv.app_id
+        )
+        LIMIT :batch_size
+    """
+
+    with db.engine.connect() as conn:
+        result = conn.execute(sa.text(query), {"batch_size": batch_size})
+        return [row[0] for row in result]
+
+
+def _count_orphaned_draft_variables() -> dict[str, Any]:
+    """
+    Count orphaned draft variables by app.
+
+    Returns:
+        Dictionary with statistics about orphaned variables
+    """
+    query = """
+        SELECT
+            wdv.app_id,
+            COUNT(*) as variable_count
+        FROM workflow_draft_variables AS wdv
+        WHERE NOT EXISTS(
+            SELECT 1 FROM apps WHERE apps.id = wdv.app_id
+        )
+        GROUP BY wdv.app_id
+        ORDER BY variable_count DESC
+    """
+
+    with db.engine.connect() as conn:
+        result = conn.execute(sa.text(query))
+        orphaned_by_app = {row[0]: row[1] for row in result}
+
+        total_orphaned = sum(orphaned_by_app.values())
+        app_count = len(orphaned_by_app)
+
+        return {
+            "total_orphaned_variables": total_orphaned,
+            "orphaned_app_count": app_count,
+            "orphaned_by_app": orphaned_by_app,
+        }
+
+
+@click.command()
+@click.option("--dry-run", is_flag=True, help="Show what would be deleted without actually deleting")
+@click.option("--batch-size", default=1000, help="Number of records to process per batch (default 1000)")
+@click.option("--max-apps", default=None, type=int, help="Maximum number of apps to process (default: no limit)")
+@click.option("-f", "--force", is_flag=True, help="Skip user confirmation and force the command to execute.")
+def cleanup_orphaned_draft_variables(
+    dry_run: bool,
+    batch_size: int,
+    max_apps: int | None,
+    force: bool = False,
+):
+    """
+    Clean up orphaned draft variables from the database.
+
+    This script finds and removes draft variables that belong to apps
+    that no longer exist in the database.
+    """
+    logger = logging.getLogger(__name__)
+
+    # Get statistics
+    stats = _count_orphaned_draft_variables()
+
+    logger.info("Found %s orphaned draft variables", stats["total_orphaned_variables"])
+    logger.info("Across %s non-existent apps", stats["orphaned_app_count"])
+
+    if stats["total_orphaned_variables"] == 0:
+        logger.info("No orphaned draft variables found. Exiting.")
+        return
+
+    if dry_run:
+        logger.info("DRY RUN: Would delete the following:")
+        for app_id, count in sorted(stats["orphaned_by_app"].items(), key=lambda x: x[1], reverse=True)[
+            :10
+        ]:  # Show top 10
+            logger.info("  App %s: %s variables", app_id, count)
+        if len(stats["orphaned_by_app"]) > 10:
+            logger.info("  ... and %s more apps", len(stats["orphaned_by_app"]) - 10)
+        return
+
+    # Confirm deletion
+    if not force:
+        click.confirm(
+            f"Are you sure you want to delete {stats['total_orphaned_variables']} "
+            f"orphaned draft variables from {stats['orphaned_app_count']} apps?",
+            abort=True,
+        )
+
+    total_deleted = 0
+    processed_apps = 0
+
+    while True:
+        if max_apps and processed_apps >= max_apps:
+            logger.info("Reached maximum app limit (%s). Stopping.", max_apps)
+            break
+
+        orphaned_app_ids = _find_orphaned_draft_variables(batch_size=10)
+        if not orphaned_app_ids:
+            logger.info("No more orphaned draft variables found.")
+            break
+
+        for app_id in orphaned_app_ids:
+            if max_apps and processed_apps >= max_apps:
+                break
+
+            try:
+                deleted_count = delete_draft_variables_batch(app_id, batch_size)
+                total_deleted += deleted_count
+                processed_apps += 1
+
+                logger.info("Deleted %s variables for app %s", deleted_count, app_id)
+
+            except Exception:
+                logger.exception("Error processing app %s", app_id)
+                continue
+
+    logger.info("Cleanup completed. Total deleted: %s variables across %s apps", total_deleted, processed_apps)

api/configs/feature/__init__.py

@@ -330,17 +330,17 @@ class HttpConfig(BaseSettings):
     def WEB_API_CORS_ALLOW_ORIGINS(self) -> list[str]:
         return self.inner_WEB_API_CORS_ALLOW_ORIGINS.split(",")
 
-    HTTP_REQUEST_MAX_CONNECT_TIMEOUT: Annotated[
-        PositiveInt, Field(ge=10, description="Maximum connection timeout in seconds for HTTP requests")
-    ] = 10
+    HTTP_REQUEST_MAX_CONNECT_TIMEOUT: int = Field(
+        ge=1, description="Maximum connection timeout in seconds for HTTP requests", default=10
+    )
 
-    HTTP_REQUEST_MAX_READ_TIMEOUT: Annotated[
-        PositiveInt, Field(ge=60, description="Maximum read timeout in seconds for HTTP requests")
-    ] = 60
+    HTTP_REQUEST_MAX_READ_TIMEOUT: int = Field(
+        ge=1, description="Maximum read timeout in seconds for HTTP requests", default=60
+    )
 
-    HTTP_REQUEST_MAX_WRITE_TIMEOUT: Annotated[
-        PositiveInt, Field(ge=10, description="Maximum write timeout in seconds for HTTP requests")
-    ] = 20
+    HTTP_REQUEST_MAX_WRITE_TIMEOUT: int = Field(
+        ge=1, description="Maximum write timeout in seconds for HTTP requests", default=20
+    )
 
     HTTP_REQUEST_NODE_MAX_BINARY_SIZE: PositiveInt = Field(
         description="Maximum allowed size in bytes for binary data in HTTP requests",
@@ -552,12 +552,18 @@ class RepositoryConfig(BaseSettings):
     """
 
     CORE_WORKFLOW_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowExecution. Specify as a module path",
+        description="Repository implementation for WorkflowExecution. Options: "
+        "'core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository' (default), "
+        "'core.repositories.celery_workflow_execution_repository.CeleryWorkflowExecutionRepository'",
         default="core.repositories.sqlalchemy_workflow_execution_repository.SQLAlchemyWorkflowExecutionRepository",
     )
 
     CORE_WORKFLOW_NODE_EXECUTION_REPOSITORY: str = Field(
-        description="Repository implementation for WorkflowNodeExecution. Specify as a module path",
+        description="Repository implementation for WorkflowNodeExecution. Options: "
+        "'core.repositories.sqlalchemy_workflow_node_execution_repository."
+        "SQLAlchemyWorkflowNodeExecutionRepository' (default), "
+        "'core.repositories.celery_workflow_node_execution_repository."
+        "CeleryWorkflowNodeExecutionRepository'",
         default="core.repositories.sqlalchemy_workflow_node_execution_repository.SQLAlchemyWorkflowNodeExecutionRepository",
     )
 

api/configs/middleware/__init__.py

@@ -10,6 +10,7 @@ from .storage.aliyun_oss_storage_config import AliyunOSSStorageConfig
 from .storage.amazon_s3_storage_config import S3StorageConfig
 from .storage.azure_blob_storage_config import AzureBlobStorageConfig
 from .storage.baidu_obs_storage_config import BaiduOBSStorageConfig
+from .storage.clickzetta_volume_storage_config import ClickZettaVolumeStorageConfig
 from .storage.google_cloud_storage_config import GoogleCloudStorageConfig
 from .storage.huawei_obs_storage_config import HuaweiCloudOBSStorageConfig
 from .storage.oci_storage_config import OCIStorageConfig
@@ -20,6 +21,7 @@ from .storage.volcengine_tos_storage_config import VolcengineTOSStorageConfig
 from .vdb.analyticdb_config import AnalyticdbConfig
 from .vdb.baidu_vector_config import BaiduVectorDBConfig
 from .vdb.chroma_config import ChromaConfig
+from .vdb.clickzetta_config import ClickzettaConfig
 from .vdb.couchbase_config import CouchbaseConfig
 from .vdb.elasticsearch_config import ElasticsearchConfig
 from .vdb.huawei_cloud_config import HuaweiCloudConfig
@@ -52,6 +54,7 @@ class StorageConfig(BaseSettings):
         "aliyun-oss",
         "azure-blob",
         "baidu-obs",
+        "clickzetta-volume",
         "google-storage",
         "huawei-obs",
         "oci-storage",
@@ -61,8 +64,9 @@ class StorageConfig(BaseSettings):
         "local",
     ] = Field(
         description="Type of storage to use."
-        " Options: 'opendal', '(deprecated) local', 's3', 'aliyun-oss', 'azure-blob', 'baidu-obs', 'google-storage', "
-        "'huawei-obs', 'oci-storage', 'tencent-cos', 'volcengine-tos', 'supabase'. Default is 'opendal'.",
+        " Options: 'opendal', '(deprecated) local', 's3', 'aliyun-oss', 'azure-blob', 'baidu-obs', "
+        "'clickzetta-volume', 'google-storage', 'huawei-obs', 'oci-storage', 'tencent-cos', "
+        "'volcengine-tos', 'supabase'. Default is 'opendal'.",
         default="opendal",
     )
 
@@ -140,7 +144,8 @@ class DatabaseConfig(BaseSettings):
         default="postgresql",
     )
 
-    @computed_field
+    @computed_field  # type: ignore[misc]
+    @property
     def SQLALCHEMY_DATABASE_URI(self) -> str:
         db_extras = (
             f"{self.DB_EXTRAS}&client_encoding={self.DB_CHARSET}" if self.DB_CHARSET else self.DB_EXTRAS
@@ -303,6 +308,7 @@ class MiddlewareConfig(
     AliyunOSSStorageConfig,
     AzureBlobStorageConfig,
     BaiduOBSStorageConfig,
+    ClickZettaVolumeStorageConfig,
     GoogleCloudStorageConfig,
     HuaweiCloudOBSStorageConfig,
     OCIStorageConfig,
@@ -315,6 +321,7 @@ class MiddlewareConfig(
     VectorStoreConfig,
     AnalyticdbConfig,
     ChromaConfig,
+    ClickzettaConfig,
     HuaweiCloudConfig,
     MilvusConfig,
     MyScaleConfig,

api/configs/middleware/cache/redis_config.py

@@ -39,6 +39,26 @@ class RedisConfig(BaseSettings):
         default=False,
     )
 
+    REDIS_SSL_CERT_REQS: str = Field(
+        description="SSL certificate requirements (CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED)",
+        default="CERT_NONE",
+    )
+
+    REDIS_SSL_CA_CERTS: Optional[str] = Field(
+        description="Path to the CA certificate file for SSL verification",
+        default=None,
+    )
+
+    REDIS_SSL_CERTFILE: Optional[str] = Field(
+        description="Path to the client certificate file for SSL authentication",
+        default=None,
+    )
+
+    REDIS_SSL_KEYFILE: Optional[str] = Field(
+        description="Path to the client private key file for SSL authentication",
+        default=None,
+    )
+
     REDIS_USE_SENTINEL: Optional[bool] = Field(
         description="Enable Redis Sentinel mode for high availability",
         default=False,

api/configs/middleware/storage/clickzetta_volume_storage_config.py

@@ -0,0 +1,65 @@
+"""ClickZetta Volume Storage Configuration"""
+
+from typing import Optional
+
+from pydantic import Field
+from pydantic_settings import BaseSettings
+
+
+class ClickZettaVolumeStorageConfig(BaseSettings):
+    """Configuration for ClickZetta Volume storage."""
+
+    CLICKZETTA_VOLUME_USERNAME: Optional[str] = Field(
+        description="Username for ClickZetta Volume authentication",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_PASSWORD: Optional[str] = Field(
+        description="Password for ClickZetta Volume authentication",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_INSTANCE: Optional[str] = Field(
+        description="ClickZetta instance identifier",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_SERVICE: str = Field(
+        description="ClickZetta service endpoint",
+        default="api.clickzetta.com",
+    )
+
+    CLICKZETTA_VOLUME_WORKSPACE: str = Field(
+        description="ClickZetta workspace name",
+        default="quick_start",
+    )
+
+    CLICKZETTA_VOLUME_VCLUSTER: str = Field(
+        description="ClickZetta virtual cluster name",
+        default="default_ap",
+    )
+
+    CLICKZETTA_VOLUME_SCHEMA: str = Field(
+        description="ClickZetta schema name",
+        default="dify",
+    )
+
+    CLICKZETTA_VOLUME_TYPE: str = Field(
+        description="ClickZetta volume type (table|user|external)",
+        default="user",
+    )
+
+    CLICKZETTA_VOLUME_NAME: Optional[str] = Field(
+        description="ClickZetta volume name for external volumes",
+        default=None,
+    )
+
+    CLICKZETTA_VOLUME_TABLE_PREFIX: str = Field(
+        description="Prefix for ClickZetta volume table names",
+        default="dataset_",
+    )
+
+    CLICKZETTA_VOLUME_DIFY_PREFIX: str = Field(
+        description="Directory prefix for User Volume to organize Dify files",
+        default="dify_km",
+    )

api/configs/middleware/vdb/clickzetta_config.py

@@ -0,0 +1,69 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class ClickzettaConfig(BaseModel):
+    """
+    Clickzetta Lakehouse vector database configuration
+    """
+
+    CLICKZETTA_USERNAME: Optional[str] = Field(
+        description="Username for authenticating with Clickzetta Lakehouse",
+        default=None,
+    )
+
+    CLICKZETTA_PASSWORD: Optional[str] = Field(
+        description="Password for authenticating with Clickzetta Lakehouse",
+        default=None,
+    )
+
+    CLICKZETTA_INSTANCE: Optional[str] = Field(
+        description="Clickzetta Lakehouse instance ID",
+        default=None,
+    )
+
+    CLICKZETTA_SERVICE: Optional[str] = Field(
+        description="Clickzetta API service endpoint (e.g., 'api.clickzetta.com')",
+        default="api.clickzetta.com",
+    )
+
+    CLICKZETTA_WORKSPACE: Optional[str] = Field(
+        description="Clickzetta workspace name",
+        default="default",
+    )
+
+    CLICKZETTA_VCLUSTER: Optional[str] = Field(
+        description="Clickzetta virtual cluster name",
+        default="default_ap",
+    )
+
+    CLICKZETTA_SCHEMA: Optional[str] = Field(
+        description="Database schema name in Clickzetta",
+        default="public",
+    )
+
+    CLICKZETTA_BATCH_SIZE: Optional[int] = Field(
+        description="Batch size for bulk insert operations",
+        default=100,
+    )
+
+    CLICKZETTA_ENABLE_INVERTED_INDEX: Optional[bool] = Field(
+        description="Enable inverted index for full-text search capabilities",
+        default=True,
+    )
+
+    CLICKZETTA_ANALYZER_TYPE: Optional[str] = Field(
+        description="Analyzer type for full-text search: keyword, english, chinese, unicode",
+        default="chinese",
+    )
+
+    CLICKZETTA_ANALYZER_MODE: Optional[str] = Field(
+        description="Analyzer mode for tokenization: max_word (fine-grained) or smart (intelligent)",
+        default="smart",
+    )
+
+    CLICKZETTA_VECTOR_DISTANCE_FUNCTION: Optional[str] = Field(
+        description="Distance function for vector similarity: l2_distance or cosine_distance",
+        default="cosine_distance",
+    )

api/constants/__init__.py

@@ -9,10 +9,10 @@ DEFAULT_FILE_NUMBER_LIMITS = 3
 IMAGE_EXTENSIONS = ["jpg", "jpeg", "png", "webp", "gif", "svg"]
 IMAGE_EXTENSIONS.extend([ext.upper() for ext in IMAGE_EXTENSIONS])
 
-VIDEO_EXTENSIONS = ["mp4", "mov", "mpeg", "mpga"]
+VIDEO_EXTENSIONS = ["mp4", "mov", "mpeg", "webm"]
 VIDEO_EXTENSIONS.extend([ext.upper() for ext in VIDEO_EXTENSIONS])
 
-AUDIO_EXTENSIONS = ["mp3", "m4a", "wav", "webm", "amr"]
+AUDIO_EXTENSIONS = ["mp3", "m4a", "wav", "amr", "mpga"]
 AUDIO_EXTENSIONS.extend([ext.upper() for ext in AUDIO_EXTENSIONS])
 
 

api/controllers/common/errors.py

@@ -1,5 +1,7 @@
 from werkzeug.exceptions import HTTPException
 
+from libs.exception import BaseHTTPException
+
 
 class FilenameNotExistsError(HTTPException):
     code = 400
@@ -9,3 +11,27 @@ class FilenameNotExistsError(HTTPException):
 class RemoteFileUploadError(HTTPException):
     code = 400
     description = "Error uploading remote file."
+
+
+class FileTooLargeError(BaseHTTPException):
+    error_code = "file_too_large"
+    description = "File size exceeded. {message}"
+    code = 413
+
+
+class UnsupportedFileTypeError(BaseHTTPException):
+    error_code = "unsupported_file_type"
+    description = "File type not allowed."
+    code = 415
+
+
+class TooManyFilesError(BaseHTTPException):
+    error_code = "too_many_files"
+    description = "Only one file is allowed."
+    code = 400
+
+
+class NoFileUploadedError(BaseHTTPException):
+    error_code = "no_file_uploaded"
+    description = "Please upload your file."
+    code = 400

api/controllers/console/__init__.py

@@ -84,6 +84,7 @@ from .datasets import (
     external,
     hit_testing,
     metadata,
+    upload_file,
     website,
 )
 

api/controllers/console/app/annotation.py

@@ -3,9 +3,8 @@ from flask_login import current_user
 from flask_restful import Resource, marshal, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden
 
+from controllers.common.errors import NoFileUploadedError, TooManyFilesError
 from controllers.console import api
-from controllers.console.app.error import NoFileUploadedError
-from controllers.console.datasets.error import TooManyFilesError
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
@@ -225,14 +224,15 @@ class AnnotationBatchImportApi(Resource):
             raise Forbidden()
 
         app_id = str(app_id)
-        # get file from request
-        file = request.files["file"]
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
 
         if len(request.files) > 1:
             raise TooManyFilesError()
+
+        # get file from request
+        file = request.files["file"]
         # check file type
         if not file.filename or not file.filename.lower().endswith(".csv"):
             raise ValueError("Invalid file type. Only CSV files are allowed")

api/controllers/console/app/app.py

@@ -28,6 +28,12 @@ from services.feature_service import FeatureService
 ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "completion"]
 
 
+def _validate_description_length(description):
+    if description and len(description) > 400:
+        raise ValueError("Description cannot exceed 400 characters.")
+    return description
+
+
 class AppListApi(Resource):
     @setup_required
     @login_required
@@ -94,7 +100,7 @@ class AppListApi(Resource):
         """Create app"""
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, location="json")
-        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("description", type=_validate_description_length, location="json")
         parser.add_argument("mode", type=str, choices=ALLOW_CREATE_APP_MODES, location="json")
         parser.add_argument("icon_type", type=str, location="json")
         parser.add_argument("icon", type=str, location="json")
@@ -146,7 +152,7 @@ class AppApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("description", type=_validate_description_length, location="json")
         parser.add_argument("icon_type", type=str, location="json")
         parser.add_argument("icon", type=str, location="json")
         parser.add_argument("icon_background", type=str, location="json")
@@ -189,7 +195,7 @@ class AppCopyApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("name", type=str, location="json")
-        parser.add_argument("description", type=str, location="json")
+        parser.add_argument("description", type=_validate_description_length, location="json")
         parser.add_argument("icon_type", type=str, location="json")
         parser.add_argument("icon", type=str, location="json")
         parser.add_argument("icon_background", type=str, location="json")

api/controllers/console/app/completion.py

@@ -1,6 +1,7 @@
 import logging
 
 import flask_login
+from flask import request
 from flask_restful import Resource, reqparse
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -24,6 +25,7 @@ from core.errors.error import (
     ProviderTokenNotInitError,
     QuotaExceededError,
 )
+from core.helper.trace_id_helper import get_external_trace_id
 from core.model_runtime.errors.invoke import InvokeError
 from libs import helper
 from libs.helper import uuid_value
@@ -115,6 +117,10 @@ class ChatMessageApi(Resource):
         streaming = args["response_mode"] != "blocking"
         args["auto_generate_name"] = False
 
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+
         account = flask_login.current_user
 
         try:

api/controllers/console/app/error.py

@@ -79,18 +79,6 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
 class DraftWorkflowNotExist(BaseHTTPException):
     error_code = "draft_workflow_not_exist"
     description = "Draft workflow need to be initialized."

api/controllers/console/app/generator.py

@@ -1,3 +1,5 @@
+from collections.abc import Sequence
+
 from flask_login import current_user
 from flask_restful import Resource, reqparse
 
@@ -10,6 +12,9 @@ from controllers.console.app.error import (
 )
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
+from core.helper.code_executor.code_node_provider import CodeNodeProvider
+from core.helper.code_executor.javascript.javascript_code_provider import JavascriptCodeProvider
+from core.helper.code_executor.python3.python3_code_provider import Python3CodeProvider
 from core.llm_generator.llm_generator import LLMGenerator
 from core.model_runtime.errors.invoke import InvokeError
 from libs.login import login_required
@@ -107,6 +112,119 @@ class RuleStructuredOutputGenerateApi(Resource):
         return structured_output
 
 
+class InstructionGenerateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("flow_id", type=str, required=True, default="", location="json")
+        parser.add_argument("node_id", type=str, required=False, default="", location="json")
+        parser.add_argument("current", type=str, required=False, default="", location="json")
+        parser.add_argument("language", type=str, required=False, default="javascript", location="json")
+        parser.add_argument("instruction", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("model_config", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("ideal_output", type=str, required=False, default="", location="json")
+        args = parser.parse_args()
+        providers: list[type[CodeNodeProvider]] = [Python3CodeProvider, JavascriptCodeProvider]
+        code_provider: type[CodeNodeProvider] | None = next(
+            (p for p in providers if p.is_accept_language(args["language"])), None
+        )
+        code_template = code_provider.get_default_code() if code_provider else ""
+        try:
+            # Generate from nothing for a workflow node
+            if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":
+                from models import App, db
+                from services.workflow_service import WorkflowService
+
+                app = db.session.query(App).filter(App.id == args["flow_id"]).first()
+                if not app:
+                    return {"error": f"app {args['flow_id']} not found"}, 400
+                workflow = WorkflowService().get_draft_workflow(app_model=app)
+                if not workflow:
+                    return {"error": f"workflow {args['flow_id']} not found"}, 400
+                nodes: Sequence = workflow.graph_dict["nodes"]
+                node = [node for node in nodes if node["id"] == args["node_id"]]
+                if len(node) == 0:
+                    return {"error": f"node {args['node_id']} not found"}, 400
+                node_type = node[0]["data"]["type"]
+                match node_type:
+                    case "llm":
+                        return LLMGenerator.generate_rule_config(
+                            current_user.current_tenant_id,
+                            instruction=args["instruction"],
+                            model_config=args["model_config"],
+                            no_variable=True,
+                        )
+                    case "agent":
+                        return LLMGenerator.generate_rule_config(
+                            current_user.current_tenant_id,
+                            instruction=args["instruction"],
+                            model_config=args["model_config"],
+                            no_variable=True,
+                        )
+                    case "code":
+                        return LLMGenerator.generate_code(
+                            tenant_id=current_user.current_tenant_id,
+                            instruction=args["instruction"],
+                            model_config=args["model_config"],
+                            code_language=args["language"],
+                        )
+                    case _:
+                        return {"error": f"invalid node type: {node_type}"}
+            if args["node_id"] == "" and args["current"] != "":  # For legacy app without a workflow
+                return LLMGenerator.instruction_modify_legacy(
+                    tenant_id=current_user.current_tenant_id,
+                    flow_id=args["flow_id"],
+                    current=args["current"],
+                    instruction=args["instruction"],
+                    model_config=args["model_config"],
+                    ideal_output=args["ideal_output"],
+                )
+            if args["node_id"] != "" and args["current"] != "":  # For workflow node
+                return LLMGenerator.instruction_modify_workflow(
+                    tenant_id=current_user.current_tenant_id,
+                    flow_id=args["flow_id"],
+                    node_id=args["node_id"],
+                    current=args["current"],
+                    instruction=args["instruction"],
+                    model_config=args["model_config"],
+                    ideal_output=args["ideal_output"],
+                )
+            return {"error": "incompatible parameters"}, 400
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+
+
+class InstructionGenerationTemplateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self) -> dict:
+        parser = reqparse.RequestParser()
+        parser.add_argument("type", type=str, required=True, default=False, location="json")
+        args = parser.parse_args()
+        match args["type"]:
+            case "prompt":
+                from core.llm_generator.prompts import INSTRUCTION_GENERATE_TEMPLATE_PROMPT
+
+                return {"data": INSTRUCTION_GENERATE_TEMPLATE_PROMPT}
+            case "code":
+                from core.llm_generator.prompts import INSTRUCTION_GENERATE_TEMPLATE_CODE
+
+                return {"data": INSTRUCTION_GENERATE_TEMPLATE_CODE}
+            case _:
+                raise ValueError(f"Invalid type: {args['type']}")
+
+
 api.add_resource(RuleGenerateApi, "/rule-generate")
 api.add_resource(RuleCodeGenerateApi, "/rule-code-generate")
 api.add_resource(RuleStructuredOutputGenerateApi, "/rule-structured-output-generate")
+api.add_resource(InstructionGenerateApi, "/instruction-generate")
+api.add_resource(InstructionGenerationTemplateApi, "/instruction-generate/template")

api/controllers/console/app/message.py

@@ -27,7 +27,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
+from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
 from services.annotation_service import AppAnnotationService
 from services.errors.conversation import ConversationNotExistsError
 from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
@@ -124,17 +124,34 @@ class MessageFeedbackApi(Resource):
         parser.add_argument("rating", type=str, choices=["like", "dislike", None], location="json")
         args = parser.parse_args()
 
-        try:
-            MessageService.create_feedback(
-                app_model=app_model,
-                message_id=str(args["message_id"]),
-                user=current_user,
-                rating=args.get("rating"),
-                content=None,
-            )
-        except MessageNotExistsError:
+        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,
+            )
+            db.session.add(feedback)
+
+        db.session.commit()
+
         return {"result": "success"}
 
 

api/controllers/console/app/statistic.py

@@ -67,7 +67,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "message_count": i.message_count})
 
@@ -176,7 +176,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "terminal_count": i.terminal_count})
 
@@ -234,7 +234,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {"date": str(i.date), "token_count": i.token_count, "total_price": i.total_price, "currency": "USD"}
@@ -310,7 +310,7 @@ ORDER BY
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {"date": str(i.date), "interactions": float(i.interactions.quantize(Decimal("0.01")))}
@@ -373,7 +373,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {
@@ -435,7 +435,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "latency": round(i.latency * 1000, 4)})
 
@@ -495,7 +495,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "tps": round(i.tokens_per_second, 4)})
 

api/controllers/console/app/workflow.py

@@ -23,6 +23,7 @@ from core.app.app_config.features.file_upload.manager import FileUploadConfigMan
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file.models import File
+from core.helper.trace_id_helper import get_external_trace_id
 from extensions.ext_database import db
 from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
@@ -185,6 +186,10 @@ class AdvancedChatDraftWorkflowRunApi(Resource):
 
         args = parser.parse_args()
 
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+
         try:
             response = AppGenerateService.generate(
                 app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.DEBUGGER, streaming=True
@@ -373,6 +378,10 @@ class DraftWorkflowRunApi(Resource):
         parser.add_argument("files", type=list, required=False, location="json")
         args = parser.parse_args()
 
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+
         try:
             response = AppGenerateService.generate(
                 app_model=app_model,

api/controllers/console/app/workflow_draft_variable.py

@@ -163,11 +163,11 @@ class WorkflowVariableCollectionApi(Resource):
             draft_var_srv = WorkflowDraftVariableService(
                 session=session,
             )
-        workflow_vars = draft_var_srv.list_variables_without_values(
-            app_id=app_model.id,
-            page=args.page,
-            limit=args.limit,
-        )
+            workflow_vars = draft_var_srv.list_variables_without_values(
+                app_id=app_model.id,
+                page=args.page,
+                limit=args.limit,
+            )
 
         return workflow_vars
 

api/controllers/console/app/workflow_statistic.py

@@ -2,6 +2,7 @@ from datetime import datetime
 from decimal import Decimal
 
 import pytz
+import sqlalchemy as sa
 from flask import jsonify
 from flask_login import current_user
 from flask_restful import Resource, reqparse
@@ -71,7 +72,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "runs": i.runs})
 
@@ -133,7 +134,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append({"date": str(i.date), "terminal_count": i.terminal_count})
 
@@ -195,7 +196,7 @@ WHERE
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {
@@ -277,7 +278,7 @@ GROUP BY
         response_data = []
 
         with db.engine.begin() as conn:
-            rs = conn.execute(db.text(sql_query), arg_dict)
+            rs = conn.execute(sa.text(sql_query), arg_dict)
             for i in rs:
                 response_data.append(
                     {"date": str(i.date), "interactions": float(i.interactions.quantize(Decimal("0.01")))}

api/controllers/console/datasets/datasets.py

@@ -41,7 +41,7 @@ def _validate_name(name):
 
 
 def _validate_description_length(description):
-    if len(description) > 400:
+    if description and len(description) > 400:
         raise ValueError("Description cannot exceed 400 characters.")
     return description
 
@@ -113,7 +113,7 @@ class DatasetListApi(Resource):
         )
         parser.add_argument(
             "description",
-            type=str,
+            type=_validate_description_length,
             nullable=True,
             required=False,
             default="",
@@ -683,6 +683,7 @@ class DatasetRetrievalSettingApi(Resource):
                 | VectorType.HUAWEI_CLOUD
                 | VectorType.TENCENT
                 | VectorType.MATRIXONE
+                | VectorType.CLICKZETTA
             ):
                 return {
                     "retrieval_method": [
@@ -731,6 +732,7 @@ class DatasetRetrievalSettingMockApi(Resource):
                 | VectorType.TENCENT
                 | VectorType.HUAWEI_CLOUD
                 | VectorType.MATRIXONE
+                | VectorType.CLICKZETTA
             ):
                 return {
                     "retrieval_method": [

api/controllers/console/datasets/datasets_document.py

@@ -642,7 +642,7 @@ class DocumentIndexingStatusApi(DocumentResource):
         return marshal(document_dict, document_status_fields)
 
 
-class DocumentDetailApi(DocumentResource):
+class DocumentApi(DocumentResource):
     METADATA_CHOICES = {"all", "only", "without"}
 
     @setup_required
@@ -730,6 +730,28 @@ class DocumentDetailApi(DocumentResource):
 
         return response, 200
 
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def delete(self, dataset_id, document_id):
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+        dataset = DatasetService.get_dataset(dataset_id)
+        if dataset is None:
+            raise NotFound("Dataset not found.")
+        # check user's model setting
+        DatasetService.check_dataset_model_setting(dataset)
+
+        document = self.get_document(dataset_id, document_id)
+
+        try:
+            DocumentService.delete_document(document)
+        except services.errors.document.DocumentIndexingError:
+            raise DocumentIndexingError("Cannot delete document during indexing.")
+
+        return {"result": "success"}, 204
+
 
 class DocumentProcessingApi(DocumentResource):
     @setup_required
@@ -768,30 +790,6 @@ class DocumentProcessingApi(DocumentResource):
         return {"result": "success"}, 200
 
 
-class DocumentDeleteApi(DocumentResource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @cloud_edition_billing_rate_limit_check("knowledge")
-    def delete(self, dataset_id, document_id):
-        dataset_id = str(dataset_id)
-        document_id = str(document_id)
-        dataset = DatasetService.get_dataset(dataset_id)
-        if dataset is None:
-            raise NotFound("Dataset not found.")
-        # check user's model setting
-        DatasetService.check_dataset_model_setting(dataset)
-
-        document = self.get_document(dataset_id, document_id)
-
-        try:
-            DocumentService.delete_document(document)
-        except services.errors.document.DocumentIndexingError:
-            raise DocumentIndexingError("Cannot delete document during indexing.")
-
-        return {"result": "success"}, 204
-
-
 class DocumentMetadataApi(DocumentResource):
     @setup_required
     @login_required
@@ -1037,11 +1035,10 @@ api.add_resource(
 api.add_resource(DocumentBatchIndexingEstimateApi, "/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-estimate")
 api.add_resource(DocumentBatchIndexingStatusApi, "/datasets/<uuid:dataset_id>/batch/<string:batch>/indexing-status")
 api.add_resource(DocumentIndexingStatusApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/indexing-status")
-api.add_resource(DocumentDetailApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
+api.add_resource(DocumentApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 api.add_resource(
     DocumentProcessingApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/<string:action>"
 )
-api.add_resource(DocumentDeleteApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 api.add_resource(DocumentMetadataApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/metadata")
 api.add_resource(DocumentStatusApi, "/datasets/<uuid:dataset_id>/documents/status/<string:action>/batch")
 api.add_resource(DocumentPauseApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/processing/pause")

api/controllers/console/datasets/error.py

@@ -1,30 +1,6 @@
 from libs.exception import BaseHTTPException
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
 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/upload_file.py

@@ -0,0 +1,62 @@
+from flask_login import current_user
+from flask_restful import Resource
+from werkzeug.exceptions import NotFound
+
+from controllers.console import api
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+)
+from core.file import helpers as file_helpers
+from extensions.ext_database import db
+from models.dataset import Dataset
+from models.model import UploadFile
+from services.dataset_service import DocumentService
+
+
+class UploadFileApi(Resource):
+    @setup_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        """Get upload file."""
+        # check dataset
+        dataset_id = str(dataset_id)
+        dataset = (
+            db.session.query(Dataset)
+            .filter(Dataset.tenant_id == current_user.current_tenant_id, Dataset.id == dataset_id)
+            .first()
+        )
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        # check document
+        document_id = str(document_id)
+        document = DocumentService.get_document(dataset.id, document_id)
+        if not document:
+            raise NotFound("Document not found.")
+        # check upload file
+        if document.data_source_type != "upload_file":
+            raise ValueError(f"Document data source type ({document.data_source_type}) is not upload_file.")
+        data_source_info = document.data_source_info_dict
+        if data_source_info and "upload_file_id" in data_source_info:
+            file_id = data_source_info["upload_file_id"]
+            upload_file = db.session.query(UploadFile).filter(UploadFile.id == file_id).first()
+            if not upload_file:
+                raise NotFound("UploadFile not found.")
+        else:
+            raise ValueError("Upload file id not found in document data source info.")
+
+        url = file_helpers.get_signed_file_url(upload_file_id=upload_file.id)
+        return {
+            "id": upload_file.id,
+            "name": upload_file.name,
+            "size": upload_file.size,
+            "extension": upload_file.extension,
+            "url": url,
+            "download_url": f"{url}&as_attachment=true",
+            "mime_type": upload_file.mime_type,
+            "created_by": upload_file.created_by,
+            "created_at": upload_file.created_at.timestamp(),
+        }, 200
+
+
+api.add_resource(UploadFileApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/upload-file")

api/controllers/console/error.py

@@ -76,30 +76,6 @@ class EmailSendIpLimitError(BaseHTTPException):
     code = 429
 
 
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
 class UnauthorizedAndForceLogout(BaseHTTPException):
     error_code = "unauthorized_and_force_logout"
     description = "Unauthorized and force logout."
@@ -127,7 +103,7 @@ class EducationActivateLimitError(BaseHTTPException):
     code = 429
 
 
-class CompilanceRateLimitError(BaseHTTPException):
-    error_code = "compilance_rate_limit"
+class ComplianceRateLimitError(BaseHTTPException):
+    error_code = "compliance_rate_limit"
     description = "Rate limit exceeded for downloading compliance report."
     code = 429

api/controllers/console/explore/installed_app.py

@@ -58,21 +58,38 @@ class InstalledAppsListApi(Resource):
         # filter out apps that user doesn't have access to
         if FeatureService.get_system_features().webapp_auth.enabled:
             user_id = current_user.id
-            res = []
             app_ids = [installed_app["app"].id for installed_app in installed_app_list]
             webapp_settings = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids)
+
+            # Pre-filter out apps without setting or with sso_verified
+            filtered_installed_apps = []
+            app_id_to_app_code = {}
+
             for installed_app in installed_app_list:
-                webapp_setting = webapp_settings.get(installed_app["app"].id)
-                if not webapp_setting:
+                app_id = installed_app["app"].id
+                webapp_setting = webapp_settings.get(app_id)
+                if not webapp_setting or webapp_setting.access_mode == "sso_verified":
                     continue
-                if webapp_setting.access_mode == "sso_verified":
-                    continue
-                app_code = AppService.get_app_code_by_id(str(installed_app["app"].id))
-                if EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(
-                    user_id=user_id,
-                    app_code=app_code,
-                ):
+                app_code = AppService.get_app_code_by_id(str(app_id))
+                app_id_to_app_code[app_id] = app_code
+                filtered_installed_apps.append(installed_app)
+
+            app_codes = list(app_id_to_app_code.values())
+
+            # Batch permission check
+            permissions = EnterpriseService.WebAppAuth.batch_is_user_allowed_to_access_webapps(
+                user_id=user_id,
+                app_codes=app_codes,
+            )
+
+            # Keep only allowed apps
+            res = []
+            for installed_app in filtered_installed_apps:
+                app_id = installed_app["app"].id
+                app_code = app_id_to_app_code[app_id]
+                if permissions.get(app_code):
                     res.append(installed_app)
+
             installed_app_list = res
             logger.debug("installed_app_list: %s, user_id: %s", installed_app_list, user_id)
 

api/controllers/console/files.py

@@ -8,7 +8,13 @@ from werkzeug.exceptions import Forbidden
 import services
 from configs import dify_config
 from constants import DOCUMENT_EXTENSIONS
-from controllers.common.errors import FilenameNotExistsError
+from controllers.common.errors import (
+    FilenameNotExistsError,
+    FileTooLargeError,
+    NoFileUploadedError,
+    TooManyFilesError,
+    UnsupportedFileTypeError,
+)
 from controllers.console.wraps import (
     account_initialization_required,
     cloud_edition_billing_resource_check,
@@ -18,13 +24,6 @@ from fields.file_fields import file_fields, upload_config_fields
 from libs.login import login_required
 from services.file_service import FileService
 
-from .error import (
-    FileTooLargeError,
-    NoFileUploadedError,
-    TooManyFilesError,
-    UnsupportedFileTypeError,
-)
-
 PREVIEW_WORDS_LIMIT = 3000
 
 
@@ -49,7 +48,6 @@ class FileApi(Resource):
     @marshal_with(file_fields)
     @cloud_edition_billing_resource_check("documents")
     def post(self):
-        file = request.files["file"]
         source_str = request.form.get("source")
         source: Literal["datasets"] | None = "datasets" if source_str == "datasets" else None
 
@@ -58,6 +56,7 @@ class FileApi(Resource):
 
         if len(request.files) > 1:
             raise TooManyFilesError()
+        file = request.files["file"]
 
         if not file.filename:
             raise FilenameNotExistsError

api/controllers/console/remote_files.py

@@ -7,18 +7,17 @@ from flask_restful import Resource, marshal_with, reqparse
 
 import services
 from controllers.common import helpers
-from controllers.common.errors import RemoteFileUploadError
+from controllers.common.errors import (
+    FileTooLargeError,
+    RemoteFileUploadError,
+    UnsupportedFileTypeError,
+)
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
 from models.account import Account
 from services.file_service import FileService
 
-from .error import (
-    FileTooLargeError,
-    UnsupportedFileTypeError,
-)
-
 
 class RemoteFileInfoApi(Resource):
     @marshal_with(remote_file_info_fields)

api/controllers/console/version.py

@@ -32,7 +32,7 @@ class VersionApi(Resource):
             return result
 
         try:
-            response = requests.get(check_update_url, {"current_version": args.get("current_version")})
+            response = requests.get(check_update_url, {"current_version": args.get("current_version")}, timeout=(3, 10))
         except Exception as error:
             logging.warning("Check update version error: %s.", str(error))
             result["version"] = args.get("current_version")

api/controllers/console/workspace/tool_providers.py

@@ -862,6 +862,10 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument("icon_type", type=str, required=True, nullable=False, location="json")
         parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json", default="")
         parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("timeout", type=float, required=False, nullable=False, location="json", default=30)
+        parser.add_argument(
+            "sse_read_timeout", type=float, required=False, nullable=False, location="json", default=300
+        )
         args = parser.parse_args()
         user = current_user
         if not is_valid_url(args["server_url"]):
@@ -876,6 +880,8 @@ class ToolProviderMCPApi(Resource):
                 icon_background=args["icon_background"],
                 user_id=user.id,
                 server_identifier=args["server_identifier"],
+                timeout=args["timeout"],
+                sse_read_timeout=args["sse_read_timeout"],
             )
         )
 
@@ -891,6 +897,8 @@ class ToolProviderMCPApi(Resource):
         parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json")
         parser.add_argument("provider_id", type=str, required=True, nullable=False, location="json")
         parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("timeout", type=float, required=False, nullable=True, location="json")
+        parser.add_argument("sse_read_timeout", type=float, required=False, nullable=True, location="json")
         args = parser.parse_args()
         if not is_valid_url(args["server_url"]):
             if "[__HIDDEN__]" in args["server_url"]:
@@ -906,6 +914,8 @@ class ToolProviderMCPApi(Resource):
             icon_type=args["icon_type"],
             icon_background=args["icon_background"],
             server_identifier=args["server_identifier"],
+            timeout=args.get("timeout"),
+            sse_read_timeout=args.get("sse_read_timeout"),
         )
         return {"result": "success"}
 

api/controllers/console/workspace/workspace.py

@@ -7,15 +7,15 @@ from sqlalchemy import select
 from werkzeug.exceptions import Unauthorized
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.console import api
-from controllers.console.admin import admin_required
-from controllers.console.datasets.error import (
+from controllers.common.errors import (
+    FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.console import api
+from controllers.console.admin import admin_required
 from controllers.console.error import AccountNotLinkTenantError
 from controllers.console.wraps import (
     account_initialization_required,
@@ -191,9 +191,6 @@ class WebappLogoWorkspaceApi(Resource):
     @account_initialization_required
     @cloud_edition_billing_resource_check("workspace_custom")
     def post(self):
-        # get file from request
-        file = request.files["file"]
-
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
@@ -201,6 +198,8 @@ class WebappLogoWorkspaceApi(Resource):
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        # get file from request
+        file = request.files["file"]
         if not file.filename:
             raise FilenameNotExistsError
 

api/controllers/files/error.py

@@ -1,7 +0,0 @@
-from libs.exception import BaseHTTPException
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415

api/controllers/files/image_preview.py

@@ -5,8 +5,8 @@ from flask_restful import Resource, reqparse
 from werkzeug.exceptions import NotFound
 
 import services
+from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import api
-from controllers.files.error import UnsupportedFileTypeError
 from services.account_service import TenantService
 from services.file_service import FileService
 

api/controllers/files/tool_files.py

@@ -4,8 +4,8 @@ from flask import Response
 from flask_restful import Resource, reqparse
 from werkzeug.exceptions import Forbidden, NotFound
 
+from controllers.common.errors import UnsupportedFileTypeError
 from controllers.files import api
-from controllers.files.error import UnsupportedFileTypeError
 from core.tools.signature import verify_tool_file_signature
 from core.tools.tool_file_manager import ToolFileManager
 from models import db as global_db

api/controllers/files/upload.py

@@ -5,11 +5,13 @@ from flask_restful import Resource, marshal_with
 from werkzeug.exceptions import Forbidden
 
 import services
+from controllers.common.errors import (
+    FileTooLargeError,
+    UnsupportedFileTypeError,
+)
 from controllers.console.wraps import setup_required
 from controllers.files import api
-from controllers.files.error import UnsupportedFileTypeError
 from controllers.inner_api.plugin.wraps import get_user
-from controllers.service_api.app.error import FileTooLargeError
 from core.file.helpers import verify_plugin_file_signature
 from core.tools.tool_file_manager import ToolFileManager
 from fields.file_fields import file_fields

api/controllers/service_api/__init__.py

@@ -6,6 +6,6 @@ bp = Blueprint("service_api", __name__, url_prefix="/v1")
 api = ExternalApi(bp)
 
 from . import index
-from .app import annotation, app, audio, completion, conversation, file, message, site, workflow
+from .app import annotation, app, audio, completion, conversation, file, file_preview, message, site, workflow
 from .dataset import dataset, document, hit_testing, metadata, segment, upload_file
 from .workspace import models

api/controllers/service_api/app/completion.py

@@ -2,7 +2,7 @@ import logging
 
 from flask import request
 from flask_restful import Resource, reqparse
-from werkzeug.exceptions import InternalServerError, NotFound
+from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
 import services
 from controllers.service_api import api
@@ -30,6 +30,7 @@ from libs import helper
 from libs.helper import uuid_value
 from models.model import App, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
+from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 
 
@@ -113,7 +114,7 @@ class ChatApi(Resource):
         parser.add_argument("conversation_id", type=uuid_value, location="json")
         parser.add_argument("retriever_from", type=str, required=False, default="dev", location="json")
         parser.add_argument("auto_generate_name", type=bool, required=False, default=True, location="json")
-
+        parser.add_argument("workflow_id", type=str, required=False, location="json")
         args = parser.parse_args()
 
         external_trace_id = get_external_trace_id(request)
@@ -128,6 +129,12 @@ class ChatApi(Resource):
             )
 
             return helper.compact_generate_response(response)
+        except WorkflowNotFoundError as ex:
+            raise NotFound(str(ex))
+        except IsDraftWorkflowError as ex:
+            raise BadRequest(str(ex))
+        except WorkflowIdFormatError as ex:
+            raise BadRequest(str(ex))
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
         except services.errors.conversation.ConversationCompletedError:

api/controllers/service_api/app/conversation.py

@@ -1,5 +1,3 @@
-import json
-
 from flask_restful import Resource, marshal_with, reqparse
 from flask_restful.inputs import int_range
 from sqlalchemy.orm import Session
@@ -136,12 +134,15 @@ class ConversationVariableDetailApi(Resource):
         variable_id = str(variable_id)
 
         parser = reqparse.RequestParser()
-        parser.add_argument("value", required=True, location="json")
+        # using lambda is for passing the already-typed value without modification
+        # if no lambda, it will be converted to string
+        # the string cannot be converted using json.loads
+        parser.add_argument("value", required=True, location="json", type=lambda x: x)
         args = parser.parse_args()
 
         try:
             return ConversationService.update_conversation_variable(
-                app_model, conversation_id, variable_id, end_user, json.loads(args["value"])
+                app_model, conversation_id, variable_id, end_user, args["value"]
             )
         except services.errors.conversation.ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")

api/controllers/service_api/app/error.py

@@ -85,25 +85,13 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
+class FileNotFoundError(BaseHTTPException):
+    error_code = "file_not_found"
+    description = "The requested file was not found."
+    code = 404
 
 
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
+class FileAccessDeniedError(BaseHTTPException):
+    error_code = "file_access_denied"
+    description = "Access to the requested file is denied."
+    code = 403

api/controllers/service_api/app/file.py

@@ -2,14 +2,14 @@ from flask import request
 from flask_restful import Resource, marshal_with
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.service_api import api
-from controllers.service_api.app.error import (
+from controllers.common.errors import (
+    FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.service_api import api
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from fields.file_fields import file_fields
 from models.model import App, EndUser
@@ -20,18 +20,17 @@ class FileApi(Resource):
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.FORM))
     @marshal_with(file_fields)
     def post(self, app_model: App, end_user: EndUser):
-        file = request.files["file"]
-
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
 
-        if not file.mimetype:
-            raise UnsupportedFileTypeError()
-
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        file = request.files["file"]
+        if not file.mimetype:
+            raise UnsupportedFileTypeError()
+
         if not file.filename:
             raise FilenameNotExistsError
 

api/controllers/service_api/app/file_preview.py

@@ -0,0 +1,186 @@
+import logging
+from urllib.parse import quote
+
+from flask import Response
+from flask_restful import Resource, reqparse
+
+from controllers.service_api import api
+from controllers.service_api.app.error import (
+    FileAccessDeniedError,
+    FileNotFoundError,
+)
+from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
+from extensions.ext_database import db
+from extensions.ext_storage import storage
+from models.model import App, EndUser, Message, MessageFile, UploadFile
+
+logger = logging.getLogger(__name__)
+
+
+class FilePreviewApi(Resource):
+    """
+    Service API File Preview endpoint
+
+    Provides secure file preview/download functionality for external API users.
+    Files can only be accessed if they belong to messages within the requesting app's context.
+    """
+
+    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY))
+    def get(self, app_model: App, end_user: EndUser, file_id: str):
+        """
+        Preview/Download a file that was uploaded via Service API
+
+        Args:
+            app_model: The authenticated app model
+            end_user: The authenticated end user (optional)
+            file_id: UUID of the file to preview
+
+        Query Parameters:
+            user: Optional user identifier
+            as_attachment: Boolean, whether to download as attachment (default: false)
+
+        Returns:
+            Stream response with file content
+
+        Raises:
+            FileNotFoundError: File does not exist
+            FileAccessDeniedError: File access denied (not owned by app)
+        """
+        file_id = str(file_id)
+
+        # Parse query parameters
+        parser = reqparse.RequestParser()
+        parser.add_argument("as_attachment", type=bool, required=False, default=False, location="args")
+        args = parser.parse_args()
+
+        # Validate file ownership and get file objects
+        message_file, upload_file = self._validate_file_ownership(file_id, app_model.id)
+
+        # Get file content generator
+        try:
+            generator = storage.load(upload_file.key, stream=True)
+        except Exception as e:
+            raise FileNotFoundError(f"Failed to load file content: {str(e)}")
+
+        # Build response with appropriate headers
+        response = self._build_file_response(generator, upload_file, args["as_attachment"])
+
+        return response
+
+    def _validate_file_ownership(self, file_id: str, app_id: str) -> tuple[MessageFile, UploadFile]:
+        """
+        Validate that the file belongs to a message within the requesting app's context
+
+        Security validations performed:
+        1. File exists in MessageFile table (was used in a conversation)
+        2. Message belongs to the requesting app
+        3. UploadFile record exists and is accessible
+        4. File tenant matches app tenant (additional security layer)
+
+        Args:
+            file_id: UUID of the file to validate
+            app_id: UUID of the requesting app
+
+        Returns:
+            Tuple of (MessageFile, UploadFile) if validation passes
+
+        Raises:
+            FileNotFoundError: File or related records not found
+            FileAccessDeniedError: File does not belong to the app's context
+        """
+        try:
+            # Input validation
+            if not file_id or not app_id:
+                raise FileAccessDeniedError("Invalid file or app identifier")
+
+            # First, find the MessageFile that references this upload file
+            message_file = db.session.query(MessageFile).where(MessageFile.upload_file_id == file_id).first()
+
+            if not message_file:
+                raise FileNotFoundError("File not found in message context")
+
+            # Get the message and verify it belongs to the requesting app
+            message = (
+                db.session.query(Message).where(Message.id == message_file.message_id, Message.app_id == app_id).first()
+            )
+
+            if not message:
+                raise FileAccessDeniedError("File access denied: not owned by requesting app")
+
+            # Get the actual upload file record
+            upload_file = db.session.query(UploadFile).where(UploadFile.id == file_id).first()
+
+            if not upload_file:
+                raise FileNotFoundError("Upload file record not found")
+
+            # Additional security: verify tenant isolation
+            app = db.session.query(App).where(App.id == app_id).first()
+            if app and upload_file.tenant_id != app.tenant_id:
+                raise FileAccessDeniedError("File access denied: tenant mismatch")
+
+            return message_file, upload_file
+
+        except (FileNotFoundError, FileAccessDeniedError):
+            # Re-raise our custom exceptions
+            raise
+        except Exception as e:
+            # Log unexpected errors for debugging
+            logger.exception(
+                "Unexpected error during file ownership validation",
+                extra={"file_id": file_id, "app_id": app_id, "error": str(e)},
+            )
+            raise FileAccessDeniedError("File access validation failed")
+
+    def _build_file_response(self, generator, upload_file: UploadFile, as_attachment: bool = False) -> Response:
+        """
+        Build Flask Response object with appropriate headers for file streaming
+
+        Args:
+            generator: File content generator from storage
+            upload_file: UploadFile database record
+            as_attachment: Whether to set Content-Disposition as attachment
+
+        Returns:
+            Flask Response object with streaming file content
+        """
+        response = Response(
+            generator,
+            mimetype=upload_file.mime_type,
+            direct_passthrough=True,
+            headers={},
+        )
+
+        # Add Content-Length if known
+        if upload_file.size and upload_file.size > 0:
+            response.headers["Content-Length"] = str(upload_file.size)
+
+        # Add Accept-Ranges header for audio/video files to support seeking
+        if upload_file.mime_type in [
+            "audio/mpeg",
+            "audio/wav",
+            "audio/mp4",
+            "audio/ogg",
+            "audio/flac",
+            "audio/aac",
+            "video/mp4",
+            "video/webm",
+            "video/quicktime",
+            "audio/x-m4a",
+        ]:
+            response.headers["Accept-Ranges"] = "bytes"
+
+        # Set Content-Disposition for downloads
+        if as_attachment and upload_file.name:
+            encoded_filename = quote(upload_file.name)
+            response.headers["Content-Disposition"] = f"attachment; filename*=UTF-8''{encoded_filename}"
+            # Override content-type for downloads to force download
+            response.headers["Content-Type"] = "application/octet-stream"
+
+        # Add caching headers for performance
+        response.headers["Cache-Control"] = "public, max-age=3600"  # Cache for 1 hour
+
+        return response
+
+
+# Register the API endpoint
+api.add_resource(FilePreviewApi, "/files/<uuid:file_id>/preview")

api/controllers/service_api/app/workflow.py

@@ -5,7 +5,7 @@ from flask import request
 from flask_restful import Resource, fields, marshal_with, reqparse
 from flask_restful.inputs import int_range
 from sqlalchemy.orm import Session, sessionmaker
-from werkzeug.exceptions import InternalServerError
+from werkzeug.exceptions import BadRequest, InternalServerError, NotFound
 
 from controllers.service_api import api
 from controllers.service_api.app.error import (
@@ -34,6 +34,7 @@ from libs.helper import TimestampField
 from models.model import App, AppMode, EndUser
 from repositories.factory import DifyAPIRepositoryFactory
 from services.app_generate_service import AppGenerateService
+from services.errors.app import IsDraftWorkflowError, WorkflowIdFormatError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_app_service import WorkflowAppService
 
@@ -120,6 +121,59 @@ class WorkflowRunApi(Resource):
             raise InternalServerError()
 
 
+class WorkflowRunByIdApi(Resource):
+    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
+    def post(self, app_model: App, end_user: EndUser, workflow_id: str):
+        """
+        Run specific workflow by ID
+        """
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode != AppMode.WORKFLOW:
+            raise NotWorkflowAppError()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
+        args = parser.parse_args()
+
+        # Add workflow_id to args for AppGenerateService
+        args["workflow_id"] = workflow_id
+
+        external_trace_id = get_external_trace_id(request)
+        if external_trace_id:
+            args["external_trace_id"] = external_trace_id
+        streaming = args.get("response_mode") == "streaming"
+
+        try:
+            response = AppGenerateService.generate(
+                app_model=app_model, user=end_user, args=args, invoke_from=InvokeFrom.SERVICE_API, streaming=streaming
+            )
+
+            return helper.compact_generate_response(response)
+        except WorkflowNotFoundError as ex:
+            raise NotFound(str(ex))
+        except IsDraftWorkflowError as ex:
+            raise BadRequest(str(ex))
+        except WorkflowIdFormatError as ex:
+            raise BadRequest(str(ex))
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logging.exception("internal server error.")
+            raise InternalServerError()
+
+
 class WorkflowTaskStopApi(Resource):
     @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
     def post(self, app_model: App, end_user: EndUser, task_id: str):
@@ -193,5 +247,6 @@ class WorkflowAppLogApi(Resource):
 
 api.add_resource(WorkflowRunApi, "/workflows/run")
 api.add_resource(WorkflowRunDetailApi, "/workflows/run/<string:workflow_run_id>")
+api.add_resource(WorkflowRunByIdApi, "/workflows/<string:workflow_id>/run")
 api.add_resource(WorkflowTaskStopApi, "/workflows/tasks/<string:task_id>/stop")
 api.add_resource(WorkflowAppLogApi, "/workflows/logs")

api/controllers/service_api/dataset/dataset.py

@@ -29,7 +29,7 @@ def _validate_name(name):
 
 
 def _validate_description_length(description):
-    if len(description) > 400:
+    if description and len(description) > 400:
         raise ValueError("Description cannot exceed 400 characters.")
     return description
 
@@ -87,7 +87,7 @@ class DatasetListApi(DatasetApiResource):
         )
         parser.add_argument(
             "description",
-            type=str,
+            type=_validate_description_length,
             nullable=True,
             required=False,
             default="",

api/controllers/service_api/dataset/document.py

@@ -6,15 +6,15 @@ from sqlalchemy import desc, select
 from werkzeug.exceptions import Forbidden, NotFound
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.service_api import api
-from controllers.service_api.app.error import (
+from controllers.common.errors import (
+    FilenameNotExistsError,
     FileTooLargeError,
     NoFileUploadedError,
-    ProviderNotInitializeError,
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.service_api import api
+from controllers.service_api.app.error import ProviderNotInitializeError
 from controllers.service_api.dataset.error import (
     ArchivedDocumentImmutableError,
     DocumentIndexingError,
@@ -234,8 +234,6 @@ class DocumentAddByFileApi(DatasetApiResource):
                 args["retrieval_model"].get("reranking_model").get("reranking_model_name"),
             )
 
-        # save file info
-        file = request.files["file"]
         # check file
         if "file" not in request.files:
             raise NoFileUploadedError()
@@ -243,6 +241,8 @@ class DocumentAddByFileApi(DatasetApiResource):
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        # save file info
+        file = request.files["file"]
         if not file.filename:
             raise FilenameNotExistsError
 
@@ -358,39 +358,6 @@ class DocumentUpdateByFileApi(DatasetApiResource):
         return documents_and_batch_fields, 200
 
 
-class DocumentDeleteApi(DatasetApiResource):
-    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
-    def delete(self, tenant_id, dataset_id, document_id):
-        """Delete document."""
-        document_id = str(document_id)
-        dataset_id = str(dataset_id)
-        tenant_id = str(tenant_id)
-
-        # get dataset info
-        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
-
-        if not dataset:
-            raise ValueError("Dataset does not exist.")
-
-        document = DocumentService.get_document(dataset.id, document_id)
-
-        # 404 if document not found
-        if document is None:
-            raise NotFound("Document Not Exists.")
-
-        # 403 if document is archived
-        if DocumentService.check_archived(document):
-            raise ArchivedDocumentImmutableError()
-
-        try:
-            # delete document
-            DocumentService.delete_document(document)
-        except services.errors.document.DocumentIndexingError:
-            raise DocumentIndexingError("Cannot delete document during indexing.")
-
-        return 204
-
-
 class DocumentListApi(DatasetApiResource):
     def get(self, tenant_id, dataset_id):
         dataset_id = str(dataset_id)
@@ -473,7 +440,7 @@ class DocumentIndexingStatusApi(DatasetApiResource):
         return data
 
 
-class DocumentDetailApi(DatasetApiResource):
+class DocumentApi(DatasetApiResource):
     METADATA_CHOICES = {"all", "only", "without"}
 
     def get(self, tenant_id, dataset_id, document_id):
@@ -567,6 +534,37 @@ class DocumentDetailApi(DatasetApiResource):
 
         return response
 
+    @cloud_edition_billing_rate_limit_check("knowledge", "dataset")
+    def delete(self, tenant_id, dataset_id, document_id):
+        """Delete document."""
+        document_id = str(document_id)
+        dataset_id = str(dataset_id)
+        tenant_id = str(tenant_id)
+
+        # get dataset info
+        dataset = db.session.query(Dataset).where(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
+
+        if not dataset:
+            raise ValueError("Dataset does not exist.")
+
+        document = DocumentService.get_document(dataset.id, document_id)
+
+        # 404 if document not found
+        if document is None:
+            raise NotFound("Document Not Exists.")
+
+        # 403 if document is archived
+        if DocumentService.check_archived(document):
+            raise ArchivedDocumentImmutableError()
+
+        try:
+            # delete document
+            DocumentService.delete_document(document)
+        except services.errors.document.DocumentIndexingError:
+            raise DocumentIndexingError("Cannot delete document during indexing.")
+
+        return 204
+
 
 api.add_resource(
     DocumentAddByTextApi,
@@ -588,7 +586,6 @@ api.add_resource(
     "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update_by_file",
     "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/update-by-file",
 )
-api.add_resource(DocumentDeleteApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
+api.add_resource(DocumentApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 api.add_resource(DocumentListApi, "/datasets/<uuid:dataset_id>/documents")
 api.add_resource(DocumentIndexingStatusApi, "/datasets/<uuid:dataset_id>/documents/<string:batch>/indexing-status")
-api.add_resource(DocumentDetailApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")

api/controllers/service_api/dataset/error.py

@@ -1,30 +1,6 @@
 from libs.exception import BaseHTTPException
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
 class DatasetNotInitializedError(BaseHTTPException):
     error_code = "dataset_not_initialized"
     description = "The dataset is still being initialized or indexing. Please wait a moment."

api/controllers/web/app.py

@@ -1,5 +1,6 @@
 from flask import request
 from flask_restful import Resource, marshal_with, reqparse
+from werkzeug.exceptions import Unauthorized
 
 from controllers.common import fields
 from controllers.web import api
@@ -75,14 +76,14 @@ class AppWebAuthPermission(Resource):
         try:
             auth_header = request.headers.get("Authorization")
             if auth_header is None:
-                raise
+                raise Unauthorized("Authorization header is missing.")
             if " " not in auth_header:
-                raise
+                raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
 
             auth_scheme, tk = auth_header.split(None, 1)
             auth_scheme = auth_scheme.lower()
             if auth_scheme != "bearer":
-                raise
+                raise Unauthorized("Authorization scheme must be 'Bearer'")
 
             decoded = PassportService().verify(tk)
             user_id = decoded.get("user_id", "visitor")

api/controllers/web/error.py

@@ -97,30 +97,6 @@ class ProviderNotSupportSpeechToTextError(BaseHTTPException):
     code = 400
 
 
-class NoFileUploadedError(BaseHTTPException):
-    error_code = "no_file_uploaded"
-    description = "Please upload your file."
-    code = 400
-
-
-class TooManyFilesError(BaseHTTPException):
-    error_code = "too_many_files"
-    description = "Only one file is allowed."
-    code = 400
-
-
-class FileTooLargeError(BaseHTTPException):
-    error_code = "file_too_large"
-    description = "File size exceeded. {message}"
-    code = 413
-
-
-class UnsupportedFileTypeError(BaseHTTPException):
-    error_code = "unsupported_file_type"
-    description = "File type not allowed."
-    code = 415
-
-
 class WebAppAuthRequiredError(BaseHTTPException):
     error_code = "web_sso_auth_required"
     description = "Web app authentication required."

api/controllers/web/files.py

@@ -2,8 +2,13 @@ from flask import request
 from flask_restful import marshal_with
 
 import services
-from controllers.common.errors import FilenameNotExistsError
-from controllers.web.error import FileTooLargeError, NoFileUploadedError, TooManyFilesError, UnsupportedFileTypeError
+from controllers.common.errors import (
+    FilenameNotExistsError,
+    FileTooLargeError,
+    NoFileUploadedError,
+    TooManyFilesError,
+    UnsupportedFileTypeError,
+)
 from controllers.web.wraps import WebApiResource
 from fields.file_fields import file_fields
 from services.file_service import FileService
@@ -12,18 +17,17 @@ from services.file_service import FileService
 class FileApi(WebApiResource):
     @marshal_with(file_fields)
     def post(self, app_model, end_user):
-        file = request.files["file"]
-        source = request.form.get("source")
-
         if "file" not in request.files:
             raise NoFileUploadedError()
 
         if len(request.files) > 1:
             raise TooManyFilesError()
 
+        file = request.files["file"]
         if not file.filename:
             raise FilenameNotExistsError
 
+        source = request.form.get("source")
         if source not in ("datasets", None):
             source = None
 

api/controllers/web/remote_files.py

@@ -5,15 +5,17 @@ from flask_restful import marshal_with, reqparse
 
 import services
 from controllers.common import helpers
-from controllers.common.errors import RemoteFileUploadError
+from controllers.common.errors import (
+    FileTooLargeError,
+    RemoteFileUploadError,
+    UnsupportedFileTypeError,
+)
 from controllers.web.wraps import WebApiResource
 from core.file import helpers as file_helpers
 from core.helper import ssrf_proxy
 from fields.file_fields import file_fields_with_signed_url, remote_file_info_fields
 from services.file_service import FileService
 
-from .error import FileTooLargeError, UnsupportedFileTypeError
-
 
 class RemoteFileInfoApi(WebApiResource):
     @marshal_with(remote_file_info_fields)

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

@@ -118,26 +118,8 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             ):
                 return
 
-            # Init conversation variables
-            stmt = select(ConversationVariable).where(
-                ConversationVariable.app_id == self.conversation.app_id,
-                ConversationVariable.conversation_id == self.conversation.id,
-            )
-            with Session(db.engine) as session:
-                db_conversation_variables = session.scalars(stmt).all()
-                if not db_conversation_variables:
-                    # Create conversation variables if they don't exist.
-                    db_conversation_variables = [
-                        ConversationVariable.from_variable(
-                            app_id=self.conversation.app_id, conversation_id=self.conversation.id, variable=variable
-                        )
-                        for variable in self._workflow.conversation_variables
-                    ]
-                    session.add_all(db_conversation_variables)
-                # Convert database entities to variables.
-                conversation_variables = [item.to_variable() for item in db_conversation_variables]
-
-                session.commit()
+            # Initialize conversation variables
+            conversation_variables = self._initialize_conversation_variables()
 
             # Create a variable pool.
             system_inputs = SystemVariable(
@@ -292,3 +274,100 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             message_id=message_id,
             trace_manager=app_generate_entity.trace_manager,
         )
+
+    def _initialize_conversation_variables(self) -> list[VariableUnion]:
+        """
+        Initialize conversation variables for the current conversation.
+
+        This method:
+        1. Loads existing variables from the database
+        2. Creates new variables if none exist
+        3. Syncs missing variables from the workflow definition
+
+        :return: List of conversation variables ready for use
+        """
+        with Session(db.engine) as session:
+            existing_variables = self._load_existing_conversation_variables(session)
+
+            if not existing_variables:
+                # First time initialization - create all variables
+                existing_variables = self._create_all_conversation_variables(session)
+            else:
+                # Check and add any missing variables from the workflow
+                existing_variables = self._sync_missing_conversation_variables(session, existing_variables)
+
+            # Convert to Variable objects for use in the workflow
+            conversation_variables = [var.to_variable() for var in existing_variables]
+
+            session.commit()
+            return cast(list[VariableUnion], conversation_variables)
+
+    def _load_existing_conversation_variables(self, session: Session) -> list[ConversationVariable]:
+        """
+        Load existing conversation variables from the database.
+
+        :param session: Database session
+        :return: List of existing conversation variables
+        """
+        stmt = select(ConversationVariable).where(
+            ConversationVariable.app_id == self.conversation.app_id,
+            ConversationVariable.conversation_id == self.conversation.id,
+        )
+        return list(session.scalars(stmt).all())
+
+    def _create_all_conversation_variables(self, session: Session) -> list[ConversationVariable]:
+        """
+        Create all conversation variables for a new conversation.
+
+        :param session: Database session
+        :return: List of created conversation variables
+        """
+        new_variables = [
+            ConversationVariable.from_variable(
+                app_id=self.conversation.app_id, conversation_id=self.conversation.id, variable=variable
+            )
+            for variable in self._workflow.conversation_variables
+        ]
+
+        if new_variables:
+            session.add_all(new_variables)
+
+        return new_variables
+
+    def _sync_missing_conversation_variables(
+        self, session: Session, existing_variables: list[ConversationVariable]
+    ) -> list[ConversationVariable]:
+        """
+        Sync missing conversation variables from the workflow definition.
+
+        This handles the case where new variables are added to a workflow
+        after conversations have already been created.
+
+        :param session: Database session
+        :param existing_variables: List of existing conversation variables
+        :return: Updated list including any newly created variables
+        """
+        # Get IDs of existing and workflow variables
+        existing_ids = {var.id for var in existing_variables}
+        workflow_variables = {var.id: var for var in self._workflow.conversation_variables}
+
+        # Find missing variable IDs
+        missing_ids = set(workflow_variables.keys()) - existing_ids
+
+        if not missing_ids:
+            return existing_variables
+
+        # Create missing variables with their default values
+        new_variables = [
+            ConversationVariable.from_variable(
+                app_id=self.conversation.app_id,
+                conversation_id=self.conversation.id,
+                variable=workflow_variables[var_id],
+            )
+            for var_id in missing_ids
+        ]
+
+        session.add_all(new_variables)
+
+        # Return combined list
+        return existing_variables + new_variables

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

@@ -74,6 +74,7 @@ from core.workflow.system_variable import SystemVariable
 from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from events.message_event import message_was_created
 from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
 from models import Conversation, EndUser, Message, MessageFile
 from models.account import Account
 from models.enums import CreatorUserRole
@@ -568,7 +569,7 @@ class AdvancedChatAppGenerateTaskPipeline:
             )
 
         yield workflow_finish_resp
-        self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
+        self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_partial_success_event(
         self,
@@ -600,7 +601,7 @@ class AdvancedChatAppGenerateTaskPipeline:
             )
 
         yield workflow_finish_resp
-        self._base_task_pipeline._queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
+        self._base_task_pipeline.queue_manager.publish(QueueAdvancedChatMessageEndEvent(), PublishFrom.TASK_PIPELINE)
 
     def _handle_workflow_failed_event(
         self,
@@ -845,7 +846,7 @@ class AdvancedChatAppGenerateTaskPipeline:
         # Initialize graph runtime state
         graph_runtime_state: Optional[GraphRuntimeState] = None
 
-        for queue_message in self._base_task_pipeline._queue_manager.listen():
+        for queue_message in self._base_task_pipeline.queue_manager.listen():
             event = queue_message.event
 
             match event:
@@ -896,6 +897,7 @@ class AdvancedChatAppGenerateTaskPipeline:
     def _save_message(self, *, session: Session, graph_runtime_state: Optional[GraphRuntimeState] = None) -> None:
         message = self._get_message(session=session)
         message.answer = self._task_state.answer
+        message.updated_at = naive_utc_now()
         message.provider_response_latency = time.perf_counter() - self._base_task_pipeline._start_at
         message.message_metadata = self._task_state.metadata.model_dump_json()
         message_files = [
@@ -959,11 +961,11 @@ class AdvancedChatAppGenerateTaskPipeline:
         if self._base_task_pipeline._output_moderation_handler:
             if self._base_task_pipeline._output_moderation_handler.should_direct_output():
                 self._task_state.answer = self._base_task_pipeline._output_moderation_handler.get_final_output()
-                self._base_task_pipeline._queue_manager.publish(
+                self._base_task_pipeline.queue_manager.publish(
                     QueueTextChunkEvent(text=self._task_state.answer), PublishFrom.TASK_PIPELINE
                 )
 
-                self._base_task_pipeline._queue_manager.publish(
+                self._base_task_pipeline.queue_manager.publish(
                     QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE
                 )
                 return True

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

@@ -140,7 +140,9 @@ class ChatAppGenerator(MessageBasedAppGenerator):
         )
 
         # get tracing instance
-        trace_manager = TraceQueueManager(app_id=app_model.id)
+        trace_manager = TraceQueueManager(
+            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        )
 
         # init application generate entity
         application_generate_entity = ChatAppGenerateEntity(

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

@@ -124,7 +124,9 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         )
 
         # get tracing instance
-        trace_manager = TraceQueueManager(app_model.id)
+        trace_manager = TraceQueueManager(
+            app_id=app_model.id, user_id=user.id if isinstance(user, Account) else user.session_id
+        )
 
         # init application generate entity
         application_generate_entity = CompletionAppGenerateEntity(

api/core/app/apps/message_based_app_queue_manager.py

@@ -6,7 +6,6 @@ from core.app.entities.queue_entities import (
     MessageQueueMessage,
     QueueAdvancedChatMessageEndEvent,
     QueueErrorEvent,
-    QueueMessage,
     QueueMessageEndEvent,
     QueueStopEvent,
 )
@@ -22,15 +21,6 @@ class MessageBasedAppQueueManager(AppQueueManager):
         self._app_mode = app_mode
         self._message_id = str(message_id)
 
-    def construct_queue_message(self, event: AppQueueEvent) -> QueueMessage:
-        return MessageQueueMessage(
-            task_id=self._task_id,
-            message_id=self._message_id,
-            conversation_id=self._conversation_id,
-            app_mode=self._app_mode,
-            event=event,
-        )
-
     def _publish(self, event: AppQueueEvent, pub_from: PublishFrom) -> None:
         """
         Publish event to queue

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

@@ -711,7 +711,7 @@ class WorkflowAppGenerateTaskPipeline:
         # Initialize graph runtime state
         graph_runtime_state = None
 
-        for queue_message in self._base_task_pipeline._queue_manager.listen():
+        for queue_message in self._base_task_pipeline.queue_manager.listen():
             event = queue_message.event
 
             match event:

api/core/app/entities/app_invoke_entities.py

@@ -9,7 +9,6 @@ from core.app.app_config.entities import EasyUIBasedAppConfig, WorkflowUIBasedAp
 from core.entities.provider_configuration import ProviderModelBundle
 from core.file import File, FileUploadConfig
 from core.model_runtime.entities.model_entities import AIModelEntity
-from core.ops.ops_trace_manager import TraceQueueManager
 
 
 class InvokeFrom(Enum):
@@ -114,7 +113,8 @@ class AppGenerateEntity(BaseModel):
     extras: dict[str, Any] = Field(default_factory=dict)
 
     # tracing instance
-    trace_manager: Optional[TraceQueueManager] = None
+    # Using Any to avoid circular import with TraceQueueManager
+    trace_manager: Optional[Any] = None
 
 
 class EasyUIBasedAppGenerateEntity(AppGenerateEntity):

api/core/app/task_pipeline/based_generate_task_pipeline.py

@@ -37,7 +37,7 @@ class BasedGenerateTaskPipeline:
         stream: bool,
     ) -> None:
         self._application_generate_entity = application_generate_entity
-        self._queue_manager = queue_manager
+        self.queue_manager = queue_manager
         self._start_at = time.perf_counter()
         self._output_moderation_handler = self._init_output_moderation()
         self._stream = stream
@@ -113,7 +113,7 @@ class BasedGenerateTaskPipeline:
                 tenant_id=app_config.tenant_id,
                 app_id=app_config.app_id,
                 rule=ModerationRule(type=sensitive_word_avoidance.type, config=sensitive_word_avoidance.config),
-                queue_manager=self._queue_manager,
+                queue_manager=self.queue_manager,
             )
         return None
 

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -57,6 +57,7 @@ from core.prompt.utils.prompt_message_util import PromptMessageUtil
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from events.message_event import message_was_created
 from extensions.ext_database import db
+from libs.datetime_utils import naive_utc_now
 from models.model import AppMode, Conversation, Message, MessageAgentThought
 
 logger = logging.getLogger(__name__)
@@ -257,7 +258,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         Process stream response.
         :return:
         """
-        for message in self._queue_manager.listen():
+        for message in self.queue_manager.listen():
             if publisher:
                 publisher.publish(message)
             event = message.event
@@ -389,6 +390,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
             if llm_result.message.content
             else ""
         )
+        message.updated_at = naive_utc_now()
         message.answer_tokens = usage.completion_tokens
         message.answer_unit_price = usage.completion_unit_price
         message.answer_price_unit = usage.completion_price_unit
@@ -499,7 +501,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
             if self._output_moderation_handler.should_direct_output():
                 # stop subscribe new token when output moderation should direct output
                 self._task_state.llm_result.message.content = self._output_moderation_handler.get_final_output()
-                self._queue_manager.publish(
+                self.queue_manager.publish(
                     QueueLLMChunkEvent(
                         chunk=LLMResultChunk(
                             model=self._task_state.llm_result.model,
@@ -513,7 +515,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
                     PublishFrom.TASK_PIPELINE,
                 )
 
-                self._queue_manager.publish(
+                self.queue_manager.publish(
                     QueueStopEvent(stopped_by=QueueStopEvent.StopBy.OUTPUT_MODERATION), PublishFrom.TASK_PIPELINE
                 )
                 return True

api/core/app/task_pipeline/message_cycle_manager.py

@@ -23,6 +23,7 @@ from core.app.entities.task_entities import (
     MessageFileStreamResponse,
     MessageReplaceStreamResponse,
     MessageStreamResponse,
+    StreamEvent,
     WorkflowTaskState,
 )
 from core.llm_generator.llm_generator import LLMGenerator
@@ -180,11 +181,15 @@ class MessageCycleManager:
         :param message_id: message id
         :return:
         """
+        message_file = db.session.query(MessageFile).filter(MessageFile.id == message_id).first()
+        event_type = StreamEvent.MESSAGE_FILE if message_file else StreamEvent.MESSAGE
+
         return MessageStreamResponse(
             task_id=self._application_generate_entity.task_id,
             id=message_id,
             answer=answer,
             from_variable_selector=from_variable_selector,
+            event=event_type,
         )
 
     def message_replace_to_stream_response(self, answer: str, reason: str = "") -> MessageReplaceStreamResponse:

api/core/entities/provider_configuration.py

@@ -843,7 +843,7 @@ class ProviderConfiguration(BaseModel):
                     continue
 
                 status = ModelStatus.ACTIVE
-                if m.model in model_setting_map:
+                if m.model_type in model_setting_map and m.model in model_setting_map[m.model_type]:
                     model_setting = model_setting_map[m.model_type][m.model]
                     if model_setting.enabled is False:
                         status = ModelStatus.DISABLED

api/core/entities/provider_entities.py

@@ -176,7 +176,7 @@ class ProviderConfig(BasicProviderConfig):
 
     scope: AppSelectorScope | ModelSelectorScope | ToolSelectorScope | None = None
     required: bool = False
-    default: Optional[Union[int, str]] = None
+    default: Optional[Union[int, str, float, bool]] = None
     options: Optional[list[Option]] = None
     label: Optional[I18nObject] = None
     help: Optional[I18nObject] = None

api/core/file/file_manager.py

@@ -32,7 +32,7 @@ def get_attr(*, file: File, attr: FileAttribute):
         case FileAttribute.TRANSFER_METHOD:
             return file.transfer_method.value
         case FileAttribute.URL:
-            return file.remote_url
+            return _to_url(file)
         case FileAttribute.EXTENSION:
             return file.extension
         case FileAttribute.RELATED_ID:

api/core/helper/code_executor/template_transformer.py

@@ -5,7 +5,7 @@ from base64 import b64encode
 from collections.abc import Mapping
 from typing import Any
 
-from core.variables.utils import SegmentJSONEncoder
+from core.variables.utils import dumps_with_segments
 
 
 class TemplateTransformer(ABC):
@@ -93,7 +93,7 @@ class TemplateTransformer(ABC):
 
     @classmethod
     def serialize_inputs(cls, inputs: Mapping[str, Any]) -> str:
-        inputs_json_str = json.dumps(inputs, ensure_ascii=False, cls=SegmentJSONEncoder).encode()
+        inputs_json_str = dumps_with_segments(inputs, ensure_ascii=False).encode()
         input_base64_encoded = b64encode(inputs_json_str).decode("utf-8")
         return input_base64_encoded
 

api/core/helper/trace_id_helper.py

@@ -16,15 +16,33 @@ def get_external_trace_id(request: Any) -> Optional[str]:
     """
     Retrieve the trace_id from the request.
 
-    Priority: header ('X-Trace-Id'), then parameters, then JSON body. Returns None if not provided or invalid.
+    Priority:
+    1. header ('X-Trace-Id')
+    2. parameters
+    3. JSON body
+    4. Current OpenTelemetry context (if enabled)
+    5. OpenTelemetry traceparent header (if present and valid)
+
+    Returns None if no valid trace_id is provided.
     """
     trace_id = request.headers.get("X-Trace-Id")
+
     if not trace_id:
         trace_id = request.args.get("trace_id")
+
     if not trace_id and getattr(request, "is_json", False):
         json_data = getattr(request, "json", None)
         if json_data:
             trace_id = json_data.get("trace_id")
+
+    if not trace_id:
+        trace_id = get_trace_id_from_otel_context()
+
+    if not trace_id:
+        traceparent = request.headers.get("traceparent")
+        if traceparent:
+            trace_id = parse_traceparent_header(traceparent)
+
     if isinstance(trace_id, str) and is_valid_trace_id(trace_id):
         return trace_id
     return None
@@ -40,3 +58,49 @@ def extract_external_trace_id_from_args(args: Mapping[str, Any]) -> dict:
     if trace_id:
         return {"external_trace_id": trace_id}
     return {}
+
+
+def get_trace_id_from_otel_context() -> Optional[str]:
+    """
+    Retrieve the current trace ID from the active OpenTelemetry trace context.
+    Returns None if:
+    1. OpenTelemetry SDK is not installed or enabled.
+    2. There is no active span or trace context.
+    """
+    try:
+        from opentelemetry.trace import SpanContext, get_current_span
+        from opentelemetry.trace.span import INVALID_TRACE_ID
+
+        span = get_current_span()
+        if not span:
+            return None
+
+        span_context: SpanContext = span.get_span_context()
+
+        if not span_context or span_context.trace_id == INVALID_TRACE_ID:
+            return None
+
+        trace_id_hex = f"{span_context.trace_id:032x}"
+        return trace_id_hex
+
+    except Exception:
+        return None
+
+
+def parse_traceparent_header(traceparent: str) -> Optional[str]:
+    """
+    Parse the `traceparent` header to extract the trace_id.
+
+    Expected format:
+        'version-trace_id-span_id-flags'
+
+    Reference:
+        W3C Trace Context Specification: https://www.w3.org/TR/trace-context/
+    """
+    try:
+        parts = traceparent.split("-")
+        if len(parts) == 4 and len(parts[1]) == 32:
+            return parts[1]
+    except Exception:
+        pass
+    return None

api/core/llm_generator/llm_generator.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import re
+from collections.abc import Sequence
 from typing import Optional, cast
 
 import json_repair
@@ -11,6 +12,8 @@ from core.llm_generator.prompts import (
     CONVERSATION_TITLE_PROMPT,
     GENERATOR_QA_PROMPT,
     JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE,
+    LLM_MODIFY_CODE_SYSTEM,
+    LLM_MODIFY_PROMPT_SYSTEM,
     PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE,
     SYSTEM_STRUCTURED_OUTPUT_GENERATE,
     WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
@@ -24,6 +27,9 @@ from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
+from core.workflow.graph_engine.entities.event import AgentLogEvent
+from models import App, Message, WorkflowNodeExecutionModel, db
 
 
 class LLMGenerator:
@@ -388,3 +394,181 @@ class LLMGenerator:
         except Exception as e:
             logging.exception("Failed to invoke LLM model, model: %s", model_config.get("name"))
             return {"output": "", "error": f"An unexpected error occurred: {str(e)}"}
+
+    @staticmethod
+    def instruction_modify_legacy(
+        tenant_id: str, flow_id: str, current: str, instruction: str, model_config: dict, ideal_output: str | None
+    ) -> dict:
+        app: App | None = db.session.query(App).filter(App.id == flow_id).first()
+        last_run: Message | None = (
+            db.session.query(Message).filter(Message.app_id == flow_id).order_by(Message.created_at.desc()).first()
+        )
+        if not last_run:
+            return LLMGenerator.__instruction_modify_common(
+                tenant_id=tenant_id,
+                model_config=model_config,
+                last_run=None,
+                current=current,
+                error_message="",
+                instruction=instruction,
+                node_type="llm",
+                ideal_output=ideal_output,
+            )
+        last_run_dict = {
+            "query": last_run.query,
+            "answer": last_run.answer,
+            "error": last_run.error,
+        }
+        return LLMGenerator.__instruction_modify_common(
+            tenant_id=tenant_id,
+            model_config=model_config,
+            last_run=last_run_dict,
+            current=current,
+            error_message=str(last_run.error),
+            instruction=instruction,
+            node_type="llm",
+            ideal_output=ideal_output,
+        )
+
+    @staticmethod
+    def instruction_modify_workflow(
+        tenant_id: str,
+        flow_id: str,
+        node_id: str,
+        current: str,
+        instruction: str,
+        model_config: dict,
+        ideal_output: str | None,
+    ) -> dict:
+        from services.workflow_service import WorkflowService
+
+        app: App | None = db.session.query(App).filter(App.id == flow_id).first()
+        if not app:
+            raise ValueError("App not found.")
+        workflow = WorkflowService().get_draft_workflow(app_model=app)
+        if not workflow:
+            raise ValueError("Workflow not found for the given app model.")
+        last_run = WorkflowService().get_node_last_run(app_model=app, workflow=workflow, node_id=node_id)
+        try:
+            node_type = cast(WorkflowNodeExecutionModel, last_run).node_type
+        except Exception:
+            try:
+                node_type = [it for it in workflow.graph_dict["graph"]["nodes"] if it["id"] == node_id][0]["data"][
+                    "type"
+                ]
+            except Exception:
+                node_type = "llm"
+
+        if not last_run:  # Node is not executed yet
+            return LLMGenerator.__instruction_modify_common(
+                tenant_id=tenant_id,
+                model_config=model_config,
+                last_run=None,
+                current=current,
+                error_message="",
+                instruction=instruction,
+                node_type=node_type,
+                ideal_output=ideal_output,
+            )
+
+        def agent_log_of(node_execution: WorkflowNodeExecutionModel) -> Sequence:
+            raw_agent_log = node_execution.execution_metadata_dict.get(WorkflowNodeExecutionMetadataKey.AGENT_LOG)
+            if not raw_agent_log:
+                return []
+            parsed: Sequence[AgentLogEvent] = json.loads(raw_agent_log)
+
+            def dict_of_event(event: AgentLogEvent) -> dict:
+                return {
+                    "status": event.status,
+                    "error": event.error,
+                    "data": event.data,
+                }
+
+            return [dict_of_event(event) for event in parsed]
+
+        last_run_dict = {
+            "inputs": last_run.inputs_dict,
+            "status": last_run.status,
+            "error": last_run.error,
+            "agent_log": agent_log_of(last_run),
+        }
+
+        return LLMGenerator.__instruction_modify_common(
+            tenant_id=tenant_id,
+            model_config=model_config,
+            last_run=last_run_dict,
+            current=current,
+            error_message=last_run.error,
+            instruction=instruction,
+            node_type=last_run.node_type,
+            ideal_output=ideal_output,
+        )
+
+    @staticmethod
+    def __instruction_modify_common(
+        tenant_id: str,
+        model_config: dict,
+        last_run: dict | None,
+        current: str | None,
+        error_message: str | None,
+        instruction: str,
+        node_type: str,
+        ideal_output: str | None,
+    ) -> dict:
+        LAST_RUN = "{{#last_run#}}"
+        CURRENT = "{{#current#}}"
+        ERROR_MESSAGE = "{{#error_message#}}"
+        injected_instruction = instruction
+        if LAST_RUN in injected_instruction:
+            injected_instruction = injected_instruction.replace(LAST_RUN, json.dumps(last_run))
+        if CURRENT in injected_instruction:
+            injected_instruction = injected_instruction.replace(CURRENT, current or "null")
+        if ERROR_MESSAGE in injected_instruction:
+            injected_instruction = injected_instruction.replace(ERROR_MESSAGE, error_message or "null")
+        model_instance = ModelManager().get_model_instance(
+            tenant_id=tenant_id,
+            model_type=ModelType.LLM,
+            provider=model_config.get("provider", ""),
+            model=model_config.get("name", ""),
+        )
+        match node_type:
+            case "llm", "agent":
+                system_prompt = LLM_MODIFY_PROMPT_SYSTEM
+            case "code":
+                system_prompt = LLM_MODIFY_CODE_SYSTEM
+            case _:
+                system_prompt = LLM_MODIFY_PROMPT_SYSTEM
+        prompt_messages = [
+            SystemPromptMessage(content=system_prompt),
+            UserPromptMessage(
+                content=json.dumps(
+                    {
+                        "current": current,
+                        "last_run": last_run,
+                        "instruction": injected_instruction,
+                        "ideal_output": ideal_output,
+                    }
+                )
+            ),
+        ]
+        model_parameters = {"temperature": 0.4}
+
+        try:
+            response = cast(
+                LLMResult,
+                model_instance.invoke_llm(
+                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
+                ),
+            )
+
+            generated_raw = cast(str, response.message.content)
+            first_brace = generated_raw.find("{")
+            last_brace = generated_raw.rfind("}")
+            return {**json.loads(generated_raw[first_brace : last_brace + 1])}
+
+        except InvokeError as e:
+            error = str(e)
+            return {"error": f"Failed to generate code. Error: {error}"}
+        except Exception as e:
+            logging.exception("Failed to invoke LLM model, model: " + json.dumps(model_config.get("name")), exc_info=e)
+            return {"error": f"An unexpected error occurred: {str(e)}"}

api/core/llm_generator/prompts.py

@@ -309,3 +309,116 @@ eg:
 Here is the JSON schema:
 {{schema}}
 """  # noqa: E501
+
+LLM_MODIFY_PROMPT_SYSTEM = """
+Both your input and output should be in JSON format.
+
+! Below is the schema for input content !
+{
+    "type": "object",
+    "description": "The user is trying to process some content with a prompt, but the output is not as expected. They hope to achieve their goal by modifying the prompt.",
+    "properties": {
+        "current": {
+            "type": "string",
+            "description": "The prompt before modification, where placeholders {{}} will be replaced with actual values for the large language model. The content in the placeholders should not be changed."
+        },
+        "last_run": {
+            "type": "object",
+            "description": "The output result from the large language model after receiving the prompt.",
+        },
+        "instruction": {
+            "type": "string",
+            "description": "User's instruction to edit the current prompt"
+        },
+        "ideal_output": {
+            "type": "string",
+            "description": "The ideal output that the user expects from the large language model after modifying the prompt. You should compare the last output with the ideal output and make changes to the prompt to achieve the goal."
+        }
+    }
+}
+! Above is the schema for input content !
+
+! Below is the schema for output content !
+{
+    "type": "object",
+    "description": "Your feedback to the user after they provide modification suggestions.",
+    "properties": {
+        "modified": {
+            "type": "string",
+            "description": "Your modified prompt. You should change the original prompt as little as possible to achieve the goal. Keep the language of prompt if not asked to change"
+        },
+        "message": {
+            "type": "string",
+            "description": "Your feedback to the user, in the user's language, explaining what you did and your thought process in text, providing sufficient emotional value to the user."
+        }
+    },
+    "required": [
+        "modified",
+        "message"
+    ]
+}
+! Above is the schema for output content !
+
+Your output must strictly follow the schema format, do not output any content outside of the JSON body.
+"""  # noqa: E501
+
+LLM_MODIFY_CODE_SYSTEM = """
+Both your input and output should be in JSON format.
+
+! Below is the schema for input content !
+{
+    "type": "object",
+    "description": "The user is trying to process some data with a code snippet, but the result is not as expected. They hope to achieve their goal by modifying the code.",
+    "properties": {
+        "current": {
+            "type": "string",
+            "description": "The code before modification."
+        },
+        "last_run": {
+            "type": "object",
+            "description": "The result of the code.",
+        },
+        "message": {
+            "type": "string",
+            "description": "User's instruction to edit the current code"
+        }
+    }
+}
+! Above is the schema for input content !
+
+! Below is the schema for output content !
+{
+    "type": "object",
+    "description": "Your feedback to the user after they provide modification suggestions.",
+    "properties": {
+        "modified": {
+            "type": "string",
+            "description": "Your modified code. You should change the original code as little as possible to achieve the goal. Keep the programming language of code if not asked to change"
+        },
+        "message": {
+            "type": "string",
+            "description": "Your feedback to the user, in the user's language, explaining what you did and your thought process in text, providing sufficient emotional value to the user."
+        }
+    },
+    "required": [
+        "modified",
+        "message"
+    ]
+}
+! Above is the schema for output content !
+
+When you are modifying the code, you should remember:
+- Do not use print, this not work in dify sandbox.
+- Do not try dangerous call like deleting files. It's PROHIBITED.
+- Do not use any library that is not built-in in with Python.
+- Get inputs from the parameters of the function and have explicit type annotations.
+- Write proper imports at the top of the code.
+- Use return statement to return the result.
+- You should return a `dict`.
+Your output must strictly follow the schema format, do not output any content outside of the JSON body.
+"""  # noqa: E501
+
+INSTRUCTION_GENERATE_TEMPLATE_PROMPT = """The output of this prompt is not as expected: {{#last_run#}}.
+You should edit the prompt according to the IDEAL OUTPUT."""
+
+INSTRUCTION_GENERATE_TEMPLATE_CODE = """Please fix the errors in the {{#error_message#}}."""

api/core/mcp/auth/auth_provider.py

@@ -10,8 +10,6 @@ from core.mcp.types import (
 from models.tools import MCPToolProvider
 from services.tools.mcp_tools_manage_service import MCPToolManageService
 
-LATEST_PROTOCOL_VERSION = "1.0"
-
 
 class OAuthClientProvider:
     mcp_provider: MCPToolProvider

api/core/mcp/client/sse_client.py

@@ -7,6 +7,7 @@ from typing import Any, TypeAlias, final
 from urllib.parse import urljoin, urlparse
 
 import httpx
+from httpx_sse import EventSource, ServerSentEvent
 from sseclient import SSEClient
 
 from core.mcp import types
@@ -37,11 +38,6 @@ WriteQueue: TypeAlias = queue.Queue[SessionMessage | Exception | None]
 StatusQueue: TypeAlias = queue.Queue[_StatusReady | _StatusError]
 
 
-def remove_request_params(url: str) -> str:
-    """Remove request parameters from URL, keeping only the path."""
-    return urljoin(url, urlparse(url).path)
-
-
 class SSETransport:
     """SSE client transport implementation."""
 
@@ -114,7 +110,7 @@ class SSETransport:
             logger.exception("Error parsing server message")
             read_queue.put(exc)
 
-    def _handle_sse_event(self, sse, read_queue: ReadQueue, status_queue: StatusQueue) -> None:
+    def _handle_sse_event(self, sse: ServerSentEvent, read_queue: ReadQueue, status_queue: StatusQueue) -> None:
         """Handle a single SSE event.
 
         Args:
@@ -130,7 +126,7 @@ class SSETransport:
             case _:
                 logger.warning("Unknown SSE event: %s", sse.event)
 
-    def sse_reader(self, event_source, read_queue: ReadQueue, status_queue: StatusQueue) -> None:
+    def sse_reader(self, event_source: EventSource, read_queue: ReadQueue, status_queue: StatusQueue) -> None:
         """Read and process SSE events.
 
         Args:
@@ -225,7 +221,7 @@ class SSETransport:
         self,
         executor: ThreadPoolExecutor,
         client: httpx.Client,
-        event_source,
+        event_source: EventSource,
     ) -> tuple[ReadQueue, WriteQueue]:
         """Establish connection and start worker threads.
 
@@ -327,7 +323,7 @@ def send_message(http_client: httpx.Client, endpoint_url: str, session_message:
         )
         response.raise_for_status()
         logger.debug("Client message sent successfully: %s", response.status_code)
-    except Exception as exc:
+    except Exception:
         logger.exception("Error sending message")
         raise
 

api/core/mcp/client/streamable_client.py

@@ -55,14 +55,10 @@ DEFAULT_QUEUE_READ_TIMEOUT = 3
 class StreamableHTTPError(Exception):
     """Base exception for StreamableHTTP transport errors."""
 
-    pass
-
 
 class ResumptionError(StreamableHTTPError):
     """Raised when resumption request is invalid."""
 
-    pass
-
 
 @dataclass
 class RequestContext:
@@ -74,7 +70,7 @@ class RequestContext:
     session_message: SessionMessage
     metadata: ClientMessageMetadata | None
     server_to_client_queue: ServerToClientQueue  # Renamed for clarity
-    sse_read_timeout: timedelta
+    sse_read_timeout: float
 
 
 class StreamableHTTPTransport:
@@ -84,8 +80,8 @@ class StreamableHTTPTransport:
         self,
         url: str,
         headers: dict[str, Any] | None = None,
-        timeout: timedelta = timedelta(seconds=30),
-        sse_read_timeout: timedelta = timedelta(seconds=60 * 5),
+        timeout: float | timedelta = 30,
+        sse_read_timeout: float | timedelta = 60 * 5,
     ) -> None:
         """Initialize the StreamableHTTP transport.
 
@@ -97,8 +93,10 @@ class StreamableHTTPTransport:
         """
         self.url = url
         self.headers = headers or {}
-        self.timeout = timeout
-        self.sse_read_timeout = sse_read_timeout
+        self.timeout = timeout.total_seconds() if isinstance(timeout, timedelta) else timeout
+        self.sse_read_timeout = (
+            sse_read_timeout.total_seconds() if isinstance(sse_read_timeout, timedelta) else sse_read_timeout
+        )
         self.session_id: str | None = None
         self.request_headers = {
             ACCEPT: f"{JSON}, {SSE}",
@@ -186,7 +184,7 @@ class StreamableHTTPTransport:
             with ssrf_proxy_sse_connect(
                 self.url,
                 headers=headers,
-                timeout=httpx.Timeout(self.timeout.seconds, read=self.sse_read_timeout.seconds),
+                timeout=httpx.Timeout(self.timeout, read=self.sse_read_timeout),
                 client=client,
                 method="GET",
             ) as event_source:
@@ -215,7 +213,7 @@ class StreamableHTTPTransport:
         with ssrf_proxy_sse_connect(
             self.url,
             headers=headers,
-            timeout=httpx.Timeout(self.timeout.seconds, read=ctx.sse_read_timeout.seconds),
+            timeout=httpx.Timeout(self.timeout, read=self.sse_read_timeout),
             client=ctx.client,
             method="GET",
         ) as event_source:
@@ -402,8 +400,8 @@ class StreamableHTTPTransport:
 def streamablehttp_client(
     url: str,
     headers: dict[str, Any] | None = None,
-    timeout: timedelta = timedelta(seconds=30),
-    sse_read_timeout: timedelta = timedelta(seconds=60 * 5),
+    timeout: float | timedelta = 30,
+    sse_read_timeout: float | timedelta = 60 * 5,
     terminate_on_close: bool = True,
 ) -> Generator[
     tuple[
@@ -436,7 +434,7 @@ def streamablehttp_client(
         try:
             with create_ssrf_proxy_mcp_http_client(
                 headers=transport.request_headers,
-                timeout=httpx.Timeout(transport.timeout.seconds, read=transport.sse_read_timeout.seconds),
+                timeout=httpx.Timeout(transport.timeout, read=transport.sse_read_timeout),
             ) as client:
                 # Define callbacks that need access to thread pool
                 def start_get_stream() -> None:

api/core/mcp/mcp_client.py

@@ -23,12 +23,18 @@ class MCPClient:
         authed: bool = True,
         authorization_code: Optional[str] = None,
         for_list: bool = False,
+        headers: Optional[dict[str, str]] = None,
+        timeout: Optional[float] = None,
+        sse_read_timeout: Optional[float] = None,
     ):
         # Initialize info
         self.provider_id = provider_id
         self.tenant_id = tenant_id
         self.client_type = "streamable"
         self.server_url = server_url
+        self.headers = headers or {}
+        self.timeout = timeout
+        self.sse_read_timeout = sse_read_timeout
 
         # Authentication info
         self.authed = authed
@@ -43,7 +49,7 @@ class MCPClient:
         self._session: Optional[ClientSession] = None
         self._streams_context: Optional[AbstractContextManager[Any]] = None
         self._session_context: Optional[ClientSession] = None
-        self.exit_stack = ExitStack()
+        self._exit_stack = ExitStack()
 
         # Whether the client has been initialized
         self._initialized = False
@@ -90,21 +96,26 @@ class MCPClient:
             headers = (
                 {"Authorization": f"{self.token.token_type.capitalize()} {self.token.access_token}"}
                 if self.authed and self.token
-                else {}
+                else self.headers
+            )
+            self._streams_context = client_factory(
+                url=self.server_url,
+                headers=headers,
+                timeout=self.timeout,
+                sse_read_timeout=self.sse_read_timeout,
             )
-            self._streams_context = client_factory(url=self.server_url, headers=headers)
             if not self._streams_context:
                 raise MCPConnectionError("Failed to create connection context")
 
             # Use exit_stack to manage context managers properly
             if method_name == "mcp":
-                read_stream, write_stream, _ = self.exit_stack.enter_context(self._streams_context)
+                read_stream, write_stream, _ = self._exit_stack.enter_context(self._streams_context)
                 streams = (read_stream, write_stream)
             else:  # sse_client
-                streams = self.exit_stack.enter_context(self._streams_context)
+                streams = self._exit_stack.enter_context(self._streams_context)
 
             self._session_context = ClientSession(*streams)
-            self._session = self.exit_stack.enter_context(self._session_context)
+            self._session = self._exit_stack.enter_context(self._session_context)
             session = cast(ClientSession, self._session)
             session.initialize()
             return
@@ -120,9 +131,6 @@ class MCPClient:
             if first_try:
                 return self.connect_server(client_factory, method_name, first_try=False)
 
-        except MCPConnectionError:
-            raise
-
     def list_tools(self) -> list[Tool]:
         """Connect to an MCP server running with SSE transport"""
         # List available tools to verify connection
@@ -142,7 +150,7 @@ class MCPClient:
         """Clean up resources"""
         try:
             # ExitStack will handle proper cleanup of all managed context managers
-            self.exit_stack.close()
+            self._exit_stack.close()
         except Exception as e:
             logging.exception("Error during cleanup")
             raise ValueError(f"Error during cleanup: {e}")

api/core/mcp/server/streamable_http.py

@@ -16,13 +16,14 @@ from extensions.ext_database import db
 from models.model import App, AppMCPServer, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 
-"""
-Apply to MCP HTTP streamable server with stateless http
-"""
 logger = logging.getLogger(__name__)
 
 
 class MCPServerStreamableHTTPRequestHandler:
+    """
+    Apply to MCP HTTP streamable server with stateless http
+    """
+
     def __init__(
         self, app: App, request: types.ClientRequest | types.ClientNotification, user_input_form: list[VariableEntity]
     ):