Merge branch 'main' into docker-env

Co-authored-by: zhuqingchao <zhuqingchao@xiaomi.com>
Co-authored-by: crazywoola <100913391+crazywoola@users.noreply.github.com>

.claude/settings.json.example

@@ -0,0 +1,19 @@
+{
+    "permissions": {
+      "allow": [],
+      "deny": []
+    },
+    "env": {
+      "__comment": "Environment variables for MCP servers. Override in .claude/settings.local.json with actual values.",
+      "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+    },
+    "enabledMcpjsonServers": [
+      "context7",
+      "sequential-thinking",
+      "github",
+      "fetch",
+      "playwright",
+      "ide"
+    ],
+    "enableAllProjectMcpServers": true
+  }

.devcontainer/post_create_command.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-npm add -g pnpm@10.15.0
+corepack enable
 cd web && pnpm install
 pipx install uv
 

.github/workflows/api-tests.yml

@@ -1,13 +1,7 @@
 name: Run Pytest
 
 on:
-  pull_request:
-    branches:
-      - main
-    paths:
-      - api/**
-      - docker/**
-      - .github/workflows/api-tests.yml
+  workflow_call:
 
 concurrency:
   group: api-tests-${{ github.head_ref || github.run_id }}
@@ -48,11 +42,7 @@ jobs:
       - name: Run Unit tests
         run: |
           uv run --project api bash dev/pytest/pytest_unit_tests.sh
-      - name: Run ty check
-        run: |
-          cd api
-          uv add --dev ty
-          uv run ty check || true
+
       - name: Run pyrefly check
         run: |
           cd api
@@ -72,15 +62,6 @@ jobs:
       - name: Run dify config tests
         run: uv run --project api dev/pytest/pytest_config_tests.py
 
-      - name: MyPy Cache
-        uses: actions/cache@v4
-        with:
-          path: api/.mypy_cache
-          key: mypy-${{ matrix.python-version }}-${{ runner.os }}-${{ hashFiles('api/uv.lock') }}
-
-      - name: Run MyPy Checks
-        run: dev/mypy-check
-
       - name: Set up dotenvs
         run: |
           cp docker/.env.example docker/.env

.github/workflows/autofix.yml

@@ -1,10 +1,7 @@
 name: autofix.ci
 on:
-  workflow_call:
   pull_request:
-    branches: [ "main" ]
-  push:
-    branches: [ "main" ]
+    branches: ["main"]
 permissions:
   contents: read
 
@@ -18,7 +15,7 @@ jobs:
       # Use uv to ensure we have the same ruff version in CI and locally.
       - uses: astral-sh/setup-uv@v6
         with:
-          python-version: "3.12" 
+          python-version: "3.12"
       - run: |
           cd api
           uv sync --dev
@@ -29,6 +26,7 @@ jobs:
       - name: ast-grep
         run: |
           uvx --from ast-grep-cli sg --pattern 'db.session.query($WHATEVER).filter($HERE)' --rewrite 'db.session.query($WHATEVER).where($HERE)' -l py --update-all
+          uvx --from ast-grep-cli sg --pattern 'session.query($WHATEVER).filter($HERE)' --rewrite 'session.query($WHATEVER).where($HERE)' -l py --update-all
       - name: mdformat
         run: |
           uvx mdformat .

.github/workflows/db-migration-test.yml

@@ -1,13 +1,7 @@
 name: DB Migration Test
 
 on:
-  pull_request:
-    branches:
-      - main
-      - plugins/beta
-    paths:
-      - api/migrations/**
-      - .github/workflows/db-migration-test.yml
+  workflow_call:
 
 concurrency:
   group: db-migration-test-${{ github.ref }}
@@ -33,6 +27,12 @@ jobs:
 
       - name: Install dependencies
         run: uv sync --project api
+      - name: Ensure Offline migration are supported
+        run: |
+          # upgrade
+          uv run --directory api flask db upgrade 'base:head' --sql
+          # downgrade
+          uv run --directory api flask db downgrade 'head:base' --sql
 
       - name: Prepare middleware env
         run: |

.github/workflows/main-ci.yml

@@ -0,0 +1,78 @@
+name: Main CI Pipeline
+
+on:
+  pull_request:
+    branches: ["main"]
+  push:
+    branches: ["main"]
+
+permissions:
+  contents: write
+  pull-requests: write
+  checks: write
+  statuses: write
+
+concurrency:
+  group: main-ci-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+jobs:
+  # Check which paths were changed to determine which tests to run
+  check-changes:
+    name: Check Changed Files
+    runs-on: ubuntu-latest
+    outputs:
+      api-changed: ${{ steps.changes.outputs.api }}
+      web-changed: ${{ steps.changes.outputs.web }}
+      vdb-changed: ${{ steps.changes.outputs.vdb }}
+      migration-changed: ${{ steps.changes.outputs.migration }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dorny/paths-filter@v3
+        id: changes
+        with:
+          filters: |
+            api:
+              - 'api/**'
+              - 'docker/**'
+              - '.github/workflows/api-tests.yml'
+            web:
+              - 'web/**'
+            vdb:
+              - 'api/core/rag/datasource/**'
+              - 'docker/**'
+              - '.github/workflows/vdb-tests.yml'
+              - 'api/uv.lock'
+              - 'api/pyproject.toml'
+            migration:
+              - 'api/migrations/**'
+              - '.github/workflows/db-migration-test.yml'
+
+  # Run tests in parallel
+  api-tests:
+    name: API Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.api-changed == 'true'
+    uses: ./.github/workflows/api-tests.yml
+
+  web-tests:
+    name: Web Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.web-changed == 'true'
+    uses: ./.github/workflows/web-tests.yml
+
+  style-check:
+    name: Style Check
+    uses: ./.github/workflows/style.yml
+
+  vdb-tests:
+    name: VDB Tests
+    needs: check-changes
+    if: needs.check-changes.outputs.vdb-changed == 'true'
+    uses: ./.github/workflows/vdb-tests.yml
+
+  db-migration-test:
+    name: DB Migration Test
+    needs: check-changes
+    if: needs.check-changes.outputs.migration-changed == 'true'
+    uses: ./.github/workflows/db-migration-test.yml

.github/workflows/style.yml

@@ -1,9 +1,7 @@
 name: Style check
 
 on:
-  pull_request:
-    branches:
-      - main
+  workflow_call:
 
 concurrency:
   group: style-${{ github.head_ref || github.run_id }}
@@ -46,21 +44,18 @@ jobs:
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv sync --project api --dev
 
-      - name: Ruff check
+      - name: Run Basedpyright Checks
         if: steps.changed-files.outputs.any_changed == 'true'
-        run: |
-          uv run --directory api ruff --version
-          uv run --directory api ruff check ./
-          uv run --directory api ruff format --check ./
+        run: dev/basedpyright-check
+
+      - name: Run Mypy Type Checks
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: uv --directory api run mypy --exclude-gitignore --exclude 'tests/' --exclude 'migrations/' --check-untyped-defs --disable-error-code=import-untyped .
 
       - name: Dotenv check
         if: steps.changed-files.outputs.any_changed == 'true'
         run: uv run --project api dotenv-linter ./api/.env.example ./web/.env.example
 
-      - name: Lint hints
-        if: failure()
-        run: echo "Please run 'dev/reformat' to fix the fixable linting errors."
-
   web-style:
     name: Web Style
     runs-on: ubuntu-latest
@@ -102,7 +97,9 @@ jobs:
       - name: Web style check
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web
-        run: pnpm run lint
+        run: |
+          pnpm run lint
+          pnpm run eslint
 
   docker-compose-template:
     name: Docker Compose Template

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

@@ -67,12 +67,22 @@ jobs:
         working-directory: ./web
         run: pnpm run auto-gen-i18n ${{ env.FILE_ARGS }}
 
+      - name: Generate i18n type definitions
+        if: env.FILES_CHANGED == 'true'
+        working-directory: ./web
+        run: pnpm run gen:i18n-types
+
       - 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.
+          commit-message: Update i18n files and type definitions based on en-US changes
+          title: 'chore: translate i18n files and update type definitions'
+          body: |
+            This PR was automatically created to update i18n files and TypeScript type definitions based on changes in en-US locale.
+            
+            **Changes included:**
+            - Updated translation files for all locales
+            - Regenerated TypeScript type definitions for type safety
           branch: chore/automated-i18n-updates

.github/workflows/vdb-tests.yml

@@ -1,15 +1,7 @@
 name: Run VDB Tests
 
 on:
-  pull_request:
-    branches:
-      - main
-    paths:
-      - api/core/rag/datasource/**
-      - docker/**
-      - .github/workflows/vdb-tests.yml
-      - api/uv.lock
-      - api/pyproject.toml
+  workflow_call:
 
 concurrency:
   group: vdb-tests-${{ github.head_ref || github.run_id }}

.github/workflows/web-tests.yml

@@ -1,11 +1,7 @@
 name: Web Tests
 
 on:
-  pull_request:
-    branches:
-      - main
-    paths:
-      - web/**
+  workflow_call:
 
 concurrency:
   group: web-tests-${{ github.head_ref || github.run_id }}
@@ -51,6 +47,11 @@ jobs:
         working-directory: ./web
         run: pnpm install --frozen-lockfile
 
+      - name: Check i18n types synchronization
+        if: steps.changed-files.outputs.any_changed == 'true'
+        working-directory: ./web
+        run: pnpm run check:i18n-types
+
       - name: Run tests
         if: steps.changed-files.outputs.any_changed == 'true'
         working-directory: ./web

.gitignore

@@ -123,10 +123,12 @@ venv.bak/
 # mkdocs documentation
 /site
 
-# mypy
+# type checking
 .mypy_cache/
 .dmypy.json
 dmypy.json
+pyrightconfig.json
+!api/pyrightconfig.json
 
 # Pyre type checker
 .pyre/
@@ -195,8 +197,8 @@ sdks/python-client/dify_client.egg-info
 .vscode/*
 !.vscode/launch.json.template
 !.vscode/README.md
-pyrightconfig.json
 api/.vscode
+web/.vscode
 # vscode Code History Extension
 .history
 
@@ -214,6 +216,13 @@ mise.toml
 # Next.js build output
 .next/
 
+# PWA generated files
+web/public/sw.js
+web/public/sw.js.map
+web/public/workbox-*.js
+web/public/workbox-*.js.map
+web/public/fallback-*.js
+
 # AI Assistant
 .roo/
 api/.env.backup

.mcp.json

@@ -0,0 +1,34 @@
+{
+    "mcpServers": {
+      "context7": {
+        "type": "http",
+        "url": "https://mcp.context7.com/mcp"
+      },
+      "sequential-thinking": {
+        "type": "stdio",
+        "command": "npx",
+        "args": ["-y", "@modelcontextprotocol/server-sequential-thinking"],
+        "env": {}
+      },
+      "github": {
+        "type": "stdio",
+        "command": "npx",
+        "args": ["-y", "@modelcontextprotocol/server-github"],
+        "env": {
+          "GITHUB_PERSONAL_ACCESS_TOKEN": "${GITHUB_PERSONAL_ACCESS_TOKEN}"
+        }
+      },
+      "fetch": {
+        "type": "stdio",
+        "command": "uvx",
+        "args": ["mcp-server-fetch"],
+        "env": {}
+      },
+      "playwright": {
+        "type": "stdio",
+        "command": "npx",
+        "args": ["-y", "@playwright/mcp@latest"],
+        "env": {}
+      }
+    }
+  }

CLAUDE.md

@@ -32,7 +32,7 @@ uv run --project api pytest tests/integration_tests/  # Integration tests
 ./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
+uv run --directory api basedpyright         # Type checking
 ```
 
 ### Frontend (Web)
@@ -86,3 +86,4 @@ pnpm test                         # Run Jest tests
 ## Project-Specific Conventions
 
 - All async tasks use Celery with Redis as broker
+- **Internationalization**: Frontend supports multiple languages with English (`web/i18n/en-US/`) as the source. All user-facing text must use i18n keys, no hardcoded strings. Edit corresponding module files in `en-US/` directory for translations.

Makefile

@@ -4,6 +4,48 @@ WEB_IMAGE=$(DOCKER_REGISTRY)/dify-web
 API_IMAGE=$(DOCKER_REGISTRY)/dify-api
 VERSION=latest
 
+# Backend Development Environment Setup
+.PHONY: dev-setup prepare-docker prepare-web prepare-api
+
+# Default dev setup target
+dev-setup: prepare-docker prepare-web prepare-api
+	@echo "✅ Backend development environment setup complete!"
+
+# Step 1: Prepare Docker middleware
+prepare-docker:
+	@echo "🐳 Setting up Docker middleware..."
+	@cp -n docker/middleware.env.example docker/middleware.env 2>/dev/null || echo "Docker middleware.env already exists"
+	@cd docker && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p dify-middlewares-dev up -d
+	@echo "✅ Docker middleware started"
+
+# Step 2: Prepare web environment
+prepare-web:
+	@echo "🌐 Setting up web environment..."
+	@cp -n web/.env.example web/.env 2>/dev/null || echo "Web .env already exists"
+	@cd web && pnpm install
+	@cd web && pnpm build
+	@echo "✅ Web environment prepared (not started)"
+
+# Step 3: Prepare API environment
+prepare-api:
+	@echo "🔧 Setting up API environment..."
+	@cp -n api/.env.example api/.env 2>/dev/null || echo "API .env already exists"
+	@cd api && uv sync --dev
+	@cd api && uv run flask db upgrade
+	@echo "✅ API environment prepared (not started)"
+
+# Clean dev environment
+dev-clean:
+	@echo "⚠️  Stopping Docker containers..."
+	@cd docker && docker compose -f docker-compose.middleware.yaml --env-file middleware.env -p dify-middlewares-dev down
+	@echo "🗑️  Removing volumes..."
+	@rm -rf docker/volumes/db
+	@rm -rf docker/volumes/redis
+	@rm -rf docker/volumes/plugin_daemon
+	@rm -rf docker/volumes/weaviate
+	@rm -rf api/storage
+	@echo "✅ Cleanup complete"
+
 # Build Docker images
 build-web:
 	@echo "Building web Docker image: $(WEB_IMAGE):$(VERSION)..."
@@ -39,5 +81,21 @@ build-push-web: build-web push-web
 build-push-all: build-all push-all
 	@echo "All Docker images have been built and pushed."
 
+# Help target
+help:
+	@echo "Development Setup Targets:"
+	@echo "  make dev-setup      - Run all setup steps for backend dev environment"
+	@echo "  make prepare-docker - Set up Docker middleware"
+	@echo "  make prepare-web    - Set up web environment"
+	@echo "  make prepare-api    - Set up API environment"
+	@echo "  make dev-clean      - Stop Docker middleware containers"
+	@echo ""
+	@echo "Docker Build Targets:"
+	@echo "  make build-web      - Build web Docker image"
+	@echo "  make build-api      - Build API Docker image"
+	@echo "  make build-all      - Build all Docker images"
+	@echo "  make push-all       - Push all Docker images"
+	@echo "  make build-push-all - Build and push all Docker images"
+
 # Phony targets
-.PHONY: build-web build-api push-web push-api build-all push-all build-push-all
+.PHONY: build-web build-api push-web push-api build-all push-all build-push-all dev-setup prepare-docker prepare-web prepare-api dev-clean help

api/.env.example

@@ -75,6 +75,7 @@ DB_PASSWORD=difyai123456
 DB_HOST=localhost
 DB_PORT=5432
 DB_DATABASE=dify
+SQLALCHEMY_POOL_PRE_PING=true
 
 # Storage configuration
 # use for store upload files, private keys...

api/README.md

@@ -108,5 +108,5 @@ uv run celery -A app.celery beat
    ../dev/reformat               # Run all formatters and linters
    uv run ruff check --fix ./    # Fix linting issues
    uv run ruff format ./         # Format code
-   uv run mypy .                 # Type checking
+   uv run basedpyright .         # Type checking
    ```

api/app_factory.py

@@ -25,6 +25,9 @@ def create_flask_app_with_configs() -> DifyApp:
         # add an unique identifier to each request
         RecyclableContextVar.increment_thread_recycles()
 
+    # Capture the decorator's return value to avoid pyright reportUnusedFunction
+    _ = before_request
+
     return dify_app
 
 

api/child_class.py

@@ -1,11 +0,0 @@
-from tests.integration_tests.utils.parent_class import ParentClass
-
-
-class ChildClass(ParentClass):
-    """Test child class for module import helper tests"""
-
-    def __init__(self, name):
-        super().__init__(name)
-
-    def get_name(self):
-        return f"Child: {self.name}"

api/commands.py

@@ -571,7 +571,7 @@ def old_metadata_migration():
         for document in documents:
             if document.doc_metadata:
                 doc_metadata = document.doc_metadata
-                for key, value in doc_metadata.items():
+                for key in doc_metadata:
                     for field in BuiltInField:
                         if field.value == key:
                             break

api/configs/middleware/__init__.py

@@ -300,8 +300,7 @@ class DatasetQueueMonitorConfig(BaseSettings):
 
 class MiddlewareConfig(
     # place the configs in alphabet order
-    CeleryConfig,
-    DatabaseConfig,
+    CeleryConfig,  # Note: CeleryConfig already inherits from DatabaseConfig
     KeywordStoreConfig,
     RedisConfig,
     # configs of storage and storage providers

api/configs/middleware/vdb/clickzetta_config.py

@@ -1,9 +1,10 @@
 from typing import Optional
 
-from pydantic import BaseModel, Field
+from pydantic import Field
+from pydantic_settings import BaseSettings
 
 
-class ClickzettaConfig(BaseModel):
+class ClickzettaConfig(BaseSettings):
     """
     Clickzetta Lakehouse vector database configuration
     """

api/configs/middleware/vdb/matrixone_config.py

@@ -1,7 +1,8 @@
-from pydantic import BaseModel, Field
+from pydantic import Field
+from pydantic_settings import BaseSettings
 
 
-class MatrixoneConfig(BaseModel):
+class MatrixoneConfig(BaseSettings):
     """Matrixone vector database configuration."""
 
     MATRIXONE_HOST: str = Field(default="localhost", description="Host address of the Matrixone server")

api/configs/packaging/__init__.py

@@ -1,6 +1,6 @@
 from pydantic import Field
 
-from configs.packaging.pyproject import PyProjectConfig, PyProjectTomlConfig
+from configs.packaging.pyproject import PyProjectTomlConfig
 
 
 class PackagingInfo(PyProjectTomlConfig):

api/configs/remote_settings_sources/apollo/client.py

@@ -4,8 +4,9 @@ import logging
 import os
 import threading
 import time
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
 from pathlib import Path
+from typing import Any
 
 from .python_3x import http_request, makedirs_wrapper
 from .utils import (
@@ -25,13 +26,13 @@ logger = logging.getLogger(__name__)
 class ApolloClient:
     def __init__(
         self,
-        config_url,
-        app_id,
-        cluster="default",
-        secret="",
-        start_hot_update=True,
-        change_listener=None,
-        _notification_map=None,
+        config_url: str,
+        app_id: str,
+        cluster: str = "default",
+        secret: str = "",
+        start_hot_update: bool = True,
+        change_listener: Callable[[str, str, str, Any], None] | None = None,
+        _notification_map: dict[str, int] | None = None,
     ):
         # Core routing parameters
         self.config_url = config_url
@@ -47,17 +48,17 @@ class ApolloClient:
         # Private control variables
         self._cycle_time = 5
         self._stopping = False
-        self._cache = {}
-        self._no_key = {}
-        self._hash = {}
+        self._cache: dict[str, dict[str, Any]] = {}
+        self._no_key: dict[str, str] = {}
+        self._hash: dict[str, str] = {}
         self._pull_timeout = 75
         self._cache_file_path = os.path.expanduser("~") + "/.dify/config/remote-settings/apollo/cache/"
-        self._long_poll_thread = None
+        self._long_poll_thread: threading.Thread | None = None
         self._change_listener = change_listener  # "add" "delete" "update"
         if _notification_map is None:
             _notification_map = {"application": -1}
         self._notification_map = _notification_map
-        self.last_release_key = None
+        self.last_release_key: str | None = None
         # Private startup method
         self._path_checker()
         if start_hot_update:
@@ -68,7 +69,7 @@ class ApolloClient:
         heartbeat.daemon = True
         heartbeat.start()
 
-    def get_json_from_net(self, namespace="application"):
+    def get_json_from_net(self, namespace: str = "application") -> dict[str, Any] | None:
         url = "{}/configs/{}/{}/{}?releaseKey={}&ip={}".format(
             self.config_url, self.app_id, self.cluster, namespace, "", self.ip
         )
@@ -88,7 +89,7 @@ class ApolloClient:
             logger.exception("an error occurred in get_json_from_net")
             return None
 
-    def get_value(self, key, default_val=None, namespace="application"):
+    def get_value(self, key: str, default_val: Any = None, namespace: str = "application") -> Any:
         try:
             # read memory configuration
             namespace_cache = self._cache.get(namespace)
@@ -104,7 +105,8 @@ class ApolloClient:
             namespace_data = self.get_json_from_net(namespace)
             val = get_value_from_dict(namespace_data, key)
             if val is not None:
-                self._update_cache_and_file(namespace_data, namespace)
+                if namespace_data is not None:
+                    self._update_cache_and_file(namespace_data, namespace)
                 return val
 
             # read the file configuration
@@ -126,23 +128,23 @@ class ApolloClient:
     # to ensure the real-time correctness of the function call.
     # If the user does not have the same default val twice
     # and the default val is used here, there may be a problem.
-    def _set_local_cache_none(self, namespace, key):
+    def _set_local_cache_none(self, namespace: str, key: str) -> None:
         no_key = no_key_cache_key(namespace, key)
         self._no_key[no_key] = key
 
-    def _start_hot_update(self):
+    def _start_hot_update(self) -> None:
         self._long_poll_thread = threading.Thread(target=self._listener)
         # When the asynchronous thread is started, the daemon thread will automatically exit
         # when the main thread is launched.
         self._long_poll_thread.daemon = True
         self._long_poll_thread.start()
 
-    def stop(self):
+    def stop(self) -> None:
         self._stopping = True
         logger.info("Stopping listener...")
 
     # Call the set callback function, and if it is abnormal, try it out
-    def _call_listener(self, namespace, old_kv, new_kv):
+    def _call_listener(self, namespace: str, old_kv: dict[str, Any] | None, new_kv: dict[str, Any] | None) -> None:
         if self._change_listener is None:
             return
         if old_kv is None:
@@ -168,12 +170,12 @@ class ApolloClient:
         except BaseException as e:
             logger.warning(str(e))
 
-    def _path_checker(self):
+    def _path_checker(self) -> None:
         if not os.path.isdir(self._cache_file_path):
             makedirs_wrapper(self._cache_file_path)
 
     # update the local cache and file cache
-    def _update_cache_and_file(self, namespace_data, namespace="application"):
+    def _update_cache_and_file(self, namespace_data: dict[str, Any], namespace: str = "application") -> None:
         # update the local cache
         self._cache[namespace] = namespace_data
         # update the file cache
@@ -187,7 +189,7 @@ class ApolloClient:
             self._hash[namespace] = new_hash
 
     # get the configuration from the local file
-    def _get_local_cache(self, namespace="application"):
+    def _get_local_cache(self, namespace: str = "application") -> dict[str, Any]:
         cache_file_path = os.path.join(self._cache_file_path, f"{self.app_id}_configuration_{namespace}.txt")
         if os.path.isfile(cache_file_path):
             with open(cache_file_path) as f:
@@ -195,8 +197,8 @@ class ApolloClient:
             return result
         return {}
 
-    def _long_poll(self):
-        notifications = []
+    def _long_poll(self) -> None:
+        notifications: list[dict[str, Any]] = []
         for key in self._cache:
             namespace_data = self._cache[key]
             notification_id = -1
@@ -236,7 +238,7 @@ class ApolloClient:
         except Exception as e:
             logger.warning(str(e))
 
-    def _get_net_and_set_local(self, namespace, n_id, call_change=False):
+    def _get_net_and_set_local(self, namespace: str, n_id: int, call_change: bool = False) -> None:
         namespace_data = self.get_json_from_net(namespace)
         if not namespace_data:
             return
@@ -248,7 +250,7 @@ class ApolloClient:
             new_kv = namespace_data.get(CONFIGURATIONS)
             self._call_listener(namespace, old_kv, new_kv)
 
-    def _listener(self):
+    def _listener(self) -> None:
         logger.info("start long_poll")
         while not self._stopping:
             self._long_poll()
@@ -266,13 +268,13 @@ class ApolloClient:
         headers["Timestamp"] = time_unix_now
         return headers
 
-    def _heart_beat(self):
+    def _heart_beat(self) -> None:
         while not self._stopping:
             for namespace in self._notification_map:
                 self._do_heart_beat(namespace)
             time.sleep(60 * 10)  # 10 minutes
 
-    def _do_heart_beat(self, namespace):
+    def _do_heart_beat(self, namespace: str) -> None:
         url = f"{self.config_url}/configs/{self.app_id}/{self.cluster}/{namespace}?ip={self.ip}"
         try:
             code, body = http_request(url, timeout=3, headers=self._sign_headers(url))
@@ -292,7 +294,7 @@ class ApolloClient:
             logger.exception("an error occurred in _do_heart_beat")
             return None
 
-    def get_all_dicts(self, namespace):
+    def get_all_dicts(self, namespace: str) -> dict[str, Any] | None:
         namespace_data = self._cache.get(namespace)
         if namespace_data is None:
             net_namespace_data = self.get_json_from_net(namespace)

api/configs/remote_settings_sources/apollo/python_3x.py

@@ -2,6 +2,8 @@ import logging
 import os
 import ssl
 import urllib.request
+from collections.abc import Mapping
+from typing import Any
 from urllib import parse
 from urllib.error import HTTPError
 
@@ -19,9 +21,9 @@ urllib.request.install_opener(opener)
 logger = logging.getLogger(__name__)
 
 
-def http_request(url, timeout, headers={}):
+def http_request(url: str, timeout: int | float, headers: Mapping[str, str] = {}) -> tuple[int, str | None]:
     try:
-        request = urllib.request.Request(url, headers=headers)
+        request = urllib.request.Request(url, headers=dict(headers))
         res = urllib.request.urlopen(request, timeout=timeout)
         body = res.read().decode("utf-8")
         return res.code, body
@@ -33,9 +35,9 @@ def http_request(url, timeout, headers={}):
         raise e
 
 
-def url_encode(params):
+def url_encode(params: dict[str, Any]) -> str:
     return parse.urlencode(params)
 
 
-def makedirs_wrapper(path):
+def makedirs_wrapper(path: str) -> None:
     os.makedirs(path, exist_ok=True)

api/configs/remote_settings_sources/apollo/utils.py

@@ -1,5 +1,6 @@
 import hashlib
 import socket
+from typing import Any
 
 from .python_3x import url_encode
 
@@ -10,7 +11,7 @@ NAMESPACE_NAME = "namespaceName"
 
 
 # add timestamps uris and keys
-def signature(timestamp, uri, secret):
+def signature(timestamp: str, uri: str, secret: str) -> str:
     import base64
     import hmac
 
@@ -19,16 +20,16 @@ def signature(timestamp, uri, secret):
     return base64.b64encode(hmac_code).decode()
 
 
-def url_encode_wrapper(params):
+def url_encode_wrapper(params: dict[str, Any]) -> str:
     return url_encode(params)
 
 
-def no_key_cache_key(namespace, key):
+def no_key_cache_key(namespace: str, key: str) -> str:
     return f"{namespace}{len(namespace)}{key}"
 
 
 # Returns whether the obtained value is obtained, and None if it does not
-def get_value_from_dict(namespace_cache, key):
+def get_value_from_dict(namespace_cache: dict[str, Any] | None, key: str) -> Any | None:
     if namespace_cache:
         kv_data = namespace_cache.get(CONFIGURATIONS)
         if kv_data is None:
@@ -38,7 +39,7 @@ def get_value_from_dict(namespace_cache, key):
     return None
 
 
-def init_ip():
+def init_ip() -> str:
     ip = ""
     s = None
     try:

api/configs/remote_settings_sources/base.py

@@ -11,5 +11,5 @@ class RemoteSettingsSource:
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         raise NotImplementedError
 
-    def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool) -> Any:
+    def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool):
         return value

api/configs/remote_settings_sources/nacos/__init__.py

@@ -11,16 +11,16 @@ logger = logging.getLogger(__name__)
 
 from configs.remote_settings_sources.base import RemoteSettingsSource
 
-from .utils import _parse_config
+from .utils import parse_config
 
 
 class NacosSettingsSource(RemoteSettingsSource):
     def __init__(self, configs: Mapping[str, Any]):
         self.configs = configs
-        self.remote_configs: dict[str, Any] = {}
+        self.remote_configs: dict[str, str] = {}
         self.async_init()
 
-    def async_init(self):
+    def async_init(self) -> None:
         data_id = os.getenv("DIFY_ENV_NACOS_DATA_ID", "dify-api-env.properties")
         group = os.getenv("DIFY_ENV_NACOS_GROUP", "nacos-dify")
         tenant = os.getenv("DIFY_ENV_NACOS_NAMESPACE", "")
@@ -29,22 +29,19 @@ class NacosSettingsSource(RemoteSettingsSource):
         try:
             content = NacosHttpClient().http_request("/nacos/v1/cs/configs", method="GET", headers={}, params=params)
             self.remote_configs = self._parse_config(content)
-        except Exception as e:
+        except Exception:
             logger.exception("[get-access-token] exception occurred")
             raise
 
-    def _parse_config(self, content: str) -> dict:
+    def _parse_config(self, content: str) -> dict[str, str]:
         if not content:
             return {}
         try:
-            return _parse_config(self, content)
+            return parse_config(content)
         except Exception as e:
             raise RuntimeError(f"Failed to parse config: {e}")
 
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
-        if not isinstance(self.remote_configs, dict):
-            raise ValueError(f"remote configs is not dict, but {type(self.remote_configs)}")
-
         field_value = self.remote_configs.get(field_name)
         if field_value is None:
             return None, field_name, False

api/configs/remote_settings_sources/nacos/http_request.py

@@ -17,20 +17,26 @@ class NacosHttpClient:
         self.ak = os.getenv("DIFY_ENV_NACOS_ACCESS_KEY")
         self.sk = os.getenv("DIFY_ENV_NACOS_SECRET_KEY")
         self.server = os.getenv("DIFY_ENV_NACOS_SERVER_ADDR", "localhost:8848")
-        self.token = None
+        self.token: str | None = None
         self.token_ttl = 18000
         self.token_expire_time: float = 0
 
-    def http_request(self, url, method="GET", headers=None, params=None):
+    def http_request(
+        self, url: str, method: str = "GET", headers: dict[str, str] | None = None, params: dict[str, str] | None = None
+    ) -> str:
+        if headers is None:
+            headers = {}
+        if params is None:
+            params = {}
         try:
             self._inject_auth_info(headers, params)
             response = requests.request(method, url="http://" + self.server + url, headers=headers, params=params)
             response.raise_for_status()
             return response.text
-        except requests.exceptions.RequestException as e:
+        except requests.RequestException as e:
             return f"Request to Nacos failed: {e}"
 
-    def _inject_auth_info(self, headers, params, module="config"):
+    def _inject_auth_info(self, headers: dict[str, str], params: dict[str, str], module: str = "config") -> None:
         headers.update({"User-Agent": "Nacos-Http-Client-In-Dify:v0.0.1"})
 
         if module == "login":
@@ -45,16 +51,17 @@ class NacosHttpClient:
             headers["timeStamp"] = ts
         if self.username and self.password:
             self.get_access_token(force_refresh=False)
-            params["accessToken"] = self.token
+            if self.token is not None:
+                params["accessToken"] = self.token
 
-    def __do_sign(self, sign_str, sk):
+    def __do_sign(self, sign_str: str, sk: str) -> str:
         return (
             base64.encodebytes(hmac.new(sk.encode(), sign_str.encode(), digestmod=hashlib.sha1).digest())
             .decode()
             .strip()
         )
 
-    def get_sign_str(self, group, tenant, ts):
+    def get_sign_str(self, group: str, tenant: str, ts: str) -> str:
         sign_str = ""
         if tenant:
             sign_str = tenant + "+"
@@ -63,7 +70,7 @@ class NacosHttpClient:
         sign_str += ts  # Directly concatenate ts without conditional checks, because the nacos auth header forced it.
         return sign_str
 
-    def get_access_token(self, force_refresh=False):
+    def get_access_token(self, force_refresh: bool = False) -> str | None:
         current_time = time.time()
         if self.token and not force_refresh and self.token_expire_time > current_time:
             return self.token
@@ -77,6 +84,7 @@ class NacosHttpClient:
             self.token = response_data.get("accessToken")
             self.token_ttl = response_data.get("tokenTtl", 18000)
             self.token_expire_time = current_time + self.token_ttl - 10
-        except Exception as e:
+            return self.token
+        except Exception:
             logger.exception("[get-access-token] exception occur")
             raise

api/configs/remote_settings_sources/nacos/utils.py

@@ -1,4 +1,4 @@
-def _parse_config(self, content: str) -> dict[str, str]:
+def parse_config(content: str) -> dict[str, str]:
     config: dict[str, str] = {}
     if not content:
         return config

api/constants/languages.py

@@ -19,6 +19,7 @@ language_timezone_mapping = {
     "fa-IR": "Asia/Tehran",
     "sl-SI": "Europe/Ljubljana",
     "th-TH": "Asia/Bangkok",
+    "id-ID": "Asia/Jakarta",
 }
 
 languages = list(language_timezone_mapping.keys())

api/controllers/console/__init__.py

@@ -70,7 +70,7 @@ from .app import (
 )
 
 # Import auth controllers
-from .auth import activate, data_source_bearer_auth, data_source_oauth, forgot_password, login, oauth
+from .auth import activate, data_source_bearer_auth, data_source_oauth, forgot_password, login, oauth, oauth_server
 
 # Import billing controllers
 from .billing import billing, compliance

api/controllers/console/admin.py

@@ -1,4 +1,6 @@
+from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import request
 from flask_restx import Resource, reqparse
@@ -6,6 +8,8 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import NotFound, Unauthorized
 
+P = ParamSpec("P")
+R = TypeVar("R")
 from configs import dify_config
 from constants.languages import supported_language
 from controllers.console import api
@@ -14,9 +18,9 @@ from extensions.ext_database import db
 from models.model import App, InstalledApp, RecommendedApp
 
 
-def admin_required(view):
+def admin_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.ADMIN_API_KEY:
             raise Unauthorized("API key is invalid.")
 
@@ -130,15 +134,19 @@ class InsertExploreAppApi(Resource):
             app.is_public = False
 
         with Session(db.engine) as session:
-            installed_apps = session.execute(
-                select(InstalledApp).where(
-                    InstalledApp.app_id == recommended_app.app_id,
-                    InstalledApp.tenant_id != InstalledApp.app_owner_tenant_id,
+            installed_apps = (
+                session.execute(
+                    select(InstalledApp).where(
+                        InstalledApp.app_id == recommended_app.app_id,
+                        InstalledApp.tenant_id != InstalledApp.app_owner_tenant_id,
+                    )
                 )
-            ).all()
+                .scalars()
+                .all()
+            )
 
-        for installed_app in installed_apps:
-            db.session.delete(installed_app)
+            for installed_app in installed_apps:
+                session.delete(installed_app)
 
         db.session.delete(recommended_app)
         db.session.commit()

api/controllers/console/apikey.py

@@ -84,10 +84,10 @@ class BaseApiKeyListResource(Resource):
             flask_restx.abort(
                 400,
                 message=f"Cannot create more than {self.max_keys} API keys for this resource type.",
-                code="max_keys_exceeded",
+                custom="max_keys_exceeded",
             )
 
-        key = ApiToken.generate_api_key(self.token_prefix, 24)
+        key = ApiToken.generate_api_key(self.token_prefix or "", 24)
         api_token = ApiToken()
         setattr(api_token, self.resource_id_field, resource_id)
         api_token.tenant_id = current_user.current_tenant_id

api/controllers/console/app/app.py

@@ -237,9 +237,14 @@ class AppExportApi(Resource):
         # Add include_secret params
         parser = reqparse.RequestParser()
         parser.add_argument("include_secret", type=inputs.boolean, default=False, location="args")
+        parser.add_argument("workflow_id", type=str, location="args")
         args = parser.parse_args()
 
-        return {"data": AppDslService.export_dsl(app_model=app_model, include_secret=args["include_secret"])}
+        return {
+            "data": AppDslService.export_dsl(
+                app_model=app_model, include_secret=args["include_secret"], workflow_id=args.get("workflow_id")
+            )
+        }
 
 
 class AppNameApi(Resource):

api/controllers/console/app/conversation.py

@@ -117,7 +117,7 @@ class CompletionConversationDetailApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
+    @get_app_model(mode=AppMode.COMPLETION)
     def delete(self, app_model, conversation_id):
         if not current_user.is_editor:
             raise Forbidden()

api/controllers/console/app/generator.py

@@ -207,7 +207,7 @@ class InstructionGenerationTemplateApi(Resource):
     @setup_required
     @login_required
     @account_initialization_required
-    def post(self) -> dict:
+    def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("type", type=str, required=True, default=False, location="json")
         args = parser.parse_args()

api/controllers/console/app/message.py

@@ -3,6 +3,7 @@ import logging
 from flask_login import current_user
 from flask_restx import Resource, fields, marshal_with, reqparse
 from flask_restx.inputs import int_range
+from sqlalchemy import exists, select
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 from controllers.console import api
@@ -94,21 +95,22 @@ class ChatMessageListApi(Resource):
                 .all()
             )
 
-        has_more = False
+        # Initialize has_more based on whether we have a full page
         if len(history_messages) == args["limit"]:
             current_page_first_message = history_messages[-1]
-            rest_count = (
-                db.session.query(Message)
-                .where(
-                    Message.conversation_id == conversation.id,
-                    Message.created_at < current_page_first_message.created_at,
-                    Message.id != current_page_first_message.id,
+            # Check if there are more messages before the current page
+            has_more = db.session.scalar(
+                select(
+                    exists().where(
+                        Message.conversation_id == conversation.id,
+                        Message.created_at < current_page_first_message.created_at,
+                        Message.id != current_page_first_message.id,
+                    )
                 )
-                .count()
             )
-
-            if rest_count > 0:
-                has_more = True
+        else:
+            # If we don't have a full page, there are no more messages
+            has_more = False
 
         history_messages = list(reversed(history_messages))
 
@@ -128,7 +130,7 @@ class MessageFeedbackApi(Resource):
 
         message_id = str(args["message_id"])
 
-        message = db.session.query(Message).filter(Message.id == message_id, Message.app_id == app_model.id).first()
+        message = db.session.query(Message).where(Message.id == message_id, Message.app_id == app_model.id).first()
 
         if not message:
             raise NotFound("Message Not Exists.")

api/controllers/console/app/workflow.py

@@ -526,7 +526,7 @@ class PublishedWorkflowApi(Resource):
             )
 
             app_model.workflow_id = workflow.id
-            db.session.commit()
+            db.session.commit()  # NOTE: this is necessary for update app_model.workflow_id
 
             workflow_created_at = TimestampField().format(workflow.created_at)
 

api/controllers/console/app/workflow_app_log.py

@@ -27,7 +27,9 @@ class WorkflowAppLogApi(Resource):
         """
         parser = reqparse.RequestParser()
         parser.add_argument("keyword", type=str, location="args")
-        parser.add_argument("status", type=str, choices=["succeeded", "failed", "stopped"], location="args")
+        parser.add_argument(
+            "status", type=str, choices=["succeeded", "failed", "stopped", "partial-succeeded"], location="args"
+        )
         parser.add_argument(
             "created_at__before", type=str, location="args", help="Filter logs created before this timestamp"
         )

api/controllers/console/app/workflow_draft_variable.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Any, NoReturn
+from typing import NoReturn
 
 from flask import Response
 from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse
@@ -29,7 +29,7 @@ from services.workflow_service import WorkflowService
 logger = logging.getLogger(__name__)
 
 
-def _convert_values_to_json_serializable_object(value: Segment) -> Any:
+def _convert_values_to_json_serializable_object(value: Segment):
     if isinstance(value, FileSegment):
         return value.value.model_dump()
     elif isinstance(value, ArrayFileSegment):
@@ -40,7 +40,7 @@ def _convert_values_to_json_serializable_object(value: Segment) -> Any:
         return value.value
 
 
-def _serialize_var_value(variable: WorkflowDraftVariable) -> Any:
+def _serialize_var_value(variable: WorkflowDraftVariable):
     value = variable.get_value()
     # create a copy of the value to avoid affecting the model cache.
     value = value.model_copy(deep=True)

api/controllers/console/auth/data_source_oauth.py

@@ -81,7 +81,7 @@ class OAuthDataSourceBinding(Resource):
                 return {"error": "Invalid code"}, 400
             try:
                 oauth_provider.get_access_token(code)
-            except requests.exceptions.HTTPError as e:
+            except requests.HTTPError as e:
                 logger.exception(
                     "An error occurred during the OAuthCallback process with %s: %s", provider, e.response.text
                 )
@@ -104,7 +104,7 @@ class OAuthDataSourceSync(Resource):
             return {"error": "Invalid provider"}, 400
         try:
             oauth_provider.sync_data_source(binding_id)
-        except requests.exceptions.HTTPError as e:
+        except requests.HTTPError as e:
             logger.exception(
                 "An error occurred during the OAuthCallback process with %s: %s", provider, e.response.text
             )

api/controllers/console/auth/login.py

@@ -130,7 +130,7 @@ class ResetPasswordSendEmailApi(Resource):
             language = "en-US"
         try:
             account = AccountService.get_user_through_email(args["email"])
-        except AccountRegisterError as are:
+        except AccountRegisterError:
             raise AccountInFreezeError()
 
         if account is None:
@@ -162,7 +162,7 @@ class EmailCodeLoginSendEmailApi(Resource):
             language = "en-US"
         try:
             account = AccountService.get_user_through_email(args["email"])
-        except AccountRegisterError as are:
+        except AccountRegisterError:
             raise AccountInFreezeError()
 
         if account is None:
@@ -200,7 +200,7 @@ class EmailCodeLoginApi(Resource):
         AccountService.revoke_email_code_login_token(args["token"])
         try:
             account = AccountService.get_user_through_email(user_email)
-        except AccountRegisterError as are:
+        except AccountRegisterError:
             raise AccountInFreezeError()
         if account:
             tenants = TenantService.get_join_tenants(account)
@@ -223,7 +223,7 @@ class EmailCodeLoginApi(Resource):
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
-            except AccountRegisterError as are:
+            except AccountRegisterError:
                 raise AccountInFreezeError()
             except WorkspacesLimitExceededError:
                 raise WorkspacesLimitExceeded()

api/controllers/console/auth/oauth.py

@@ -80,7 +80,7 @@ class OAuthCallback(Resource):
         try:
             token = oauth_provider.get_access_token(code)
             user_info = oauth_provider.get_user_info(token)
-        except requests.exceptions.RequestException as e:
+        except requests.RequestException as e:
             error_text = e.response.text if e.response else str(e)
             logger.exception("An error occurred during the OAuth process with %s: %s", provider, error_text)
             return {"error": "OAuth process failed"}, 400

api/controllers/console/auth/oauth_server.py

@@ -0,0 +1,202 @@
+from collections.abc import Callable
+from functools import wraps
+from typing import Concatenate, ParamSpec, TypeVar, cast
+
+import flask_login
+from flask import jsonify, request
+from flask_restx import Resource, reqparse
+from werkzeug.exceptions import BadRequest, NotFound
+
+from controllers.console.wraps import account_initialization_required, setup_required
+from core.model_runtime.utils.encoders import jsonable_encoder
+from libs.login import login_required
+from models.account import Account
+from models.model import OAuthProviderApp
+from services.oauth_server import OAUTH_ACCESS_TOKEN_EXPIRES_IN, OAuthGrantType, OAuthServerService
+
+from .. import api
+
+P = ParamSpec("P")
+R = TypeVar("R")
+T = TypeVar("T")
+
+
+def oauth_server_client_id_required(view: Callable[Concatenate[T, OAuthProviderApp, P], R]):
+    @wraps(view)
+    def decorated(self: T, *args: P.args, **kwargs: P.kwargs):
+        parser = reqparse.RequestParser()
+        parser.add_argument("client_id", type=str, required=True, location="json")
+        parsed_args = parser.parse_args()
+        client_id = parsed_args.get("client_id")
+        if not client_id:
+            raise BadRequest("client_id is required")
+
+        oauth_provider_app = OAuthServerService.get_oauth_provider_app(client_id)
+        if not oauth_provider_app:
+            raise NotFound("client_id is invalid")
+
+        return view(self, oauth_provider_app, *args, **kwargs)
+
+    return decorated
+
+
+def oauth_server_access_token_required(view: Callable[Concatenate[T, OAuthProviderApp, Account, P], R]):
+    @wraps(view)
+    def decorated(self: T, oauth_provider_app: OAuthProviderApp, *args: P.args, **kwargs: P.kwargs):
+        if not isinstance(oauth_provider_app, OAuthProviderApp):
+            raise BadRequest("Invalid oauth_provider_app")
+
+        authorization_header = request.headers.get("Authorization")
+        if not authorization_header:
+            response = jsonify({"error": "Authorization header is required"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
+
+        parts = authorization_header.strip().split(None, 1)
+        if len(parts) != 2:
+            response = jsonify({"error": "Invalid Authorization header format"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
+
+        token_type = parts[0].strip()
+        if token_type.lower() != "bearer":
+            response = jsonify({"error": "token_type is invalid"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
+
+        access_token = parts[1].strip()
+        if not access_token:
+            response = jsonify({"error": "access_token is required"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
+
+        account = OAuthServerService.validate_oauth_access_token(oauth_provider_app.client_id, access_token)
+        if not account:
+            response = jsonify({"error": "access_token or client_id is invalid"})
+            response.status_code = 401
+            response.headers["WWW-Authenticate"] = "Bearer"
+            return response
+
+        return view(self, oauth_provider_app, account, *args, **kwargs)
+
+    return decorated
+
+
+class OAuthServerAppApi(Resource):
+    @setup_required
+    @oauth_server_client_id_required
+    def post(self, oauth_provider_app: OAuthProviderApp):
+        parser = reqparse.RequestParser()
+        parser.add_argument("redirect_uri", type=str, required=True, location="json")
+        parsed_args = parser.parse_args()
+        redirect_uri = parsed_args.get("redirect_uri")
+
+        # check if redirect_uri is valid
+        if redirect_uri not in oauth_provider_app.redirect_uris:
+            raise BadRequest("redirect_uri is invalid")
+
+        return jsonable_encoder(
+            {
+                "app_icon": oauth_provider_app.app_icon,
+                "app_label": oauth_provider_app.app_label,
+                "scope": oauth_provider_app.scope,
+            }
+        )
+
+
+class OAuthServerUserAuthorizeApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @oauth_server_client_id_required
+    def post(self, oauth_provider_app: OAuthProviderApp):
+        account = cast(Account, flask_login.current_user)
+        user_account_id = account.id
+
+        code = OAuthServerService.sign_oauth_authorization_code(oauth_provider_app.client_id, user_account_id)
+        return jsonable_encoder(
+            {
+                "code": code,
+            }
+        )
+
+
+class OAuthServerUserTokenApi(Resource):
+    @setup_required
+    @oauth_server_client_id_required
+    def post(self, oauth_provider_app: OAuthProviderApp):
+        parser = reqparse.RequestParser()
+        parser.add_argument("grant_type", type=str, required=True, location="json")
+        parser.add_argument("code", type=str, required=False, location="json")
+        parser.add_argument("client_secret", type=str, required=False, location="json")
+        parser.add_argument("redirect_uri", type=str, required=False, location="json")
+        parser.add_argument("refresh_token", type=str, required=False, location="json")
+        parsed_args = parser.parse_args()
+
+        try:
+            grant_type = OAuthGrantType(parsed_args["grant_type"])
+        except ValueError:
+            raise BadRequest("invalid grant_type")
+
+        if grant_type == OAuthGrantType.AUTHORIZATION_CODE:
+            if not parsed_args["code"]:
+                raise BadRequest("code is required")
+
+            if parsed_args["client_secret"] != oauth_provider_app.client_secret:
+                raise BadRequest("client_secret is invalid")
+
+            if parsed_args["redirect_uri"] not in oauth_provider_app.redirect_uris:
+                raise BadRequest("redirect_uri is invalid")
+
+            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                grant_type, code=parsed_args["code"], client_id=oauth_provider_app.client_id
+            )
+            return jsonable_encoder(
+                {
+                    "access_token": access_token,
+                    "token_type": "Bearer",
+                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                    "refresh_token": refresh_token,
+                }
+            )
+        elif grant_type == OAuthGrantType.REFRESH_TOKEN:
+            if not parsed_args["refresh_token"]:
+                raise BadRequest("refresh_token is required")
+
+            access_token, refresh_token = OAuthServerService.sign_oauth_access_token(
+                grant_type, refresh_token=parsed_args["refresh_token"], client_id=oauth_provider_app.client_id
+            )
+            return jsonable_encoder(
+                {
+                    "access_token": access_token,
+                    "token_type": "Bearer",
+                    "expires_in": OAUTH_ACCESS_TOKEN_EXPIRES_IN,
+                    "refresh_token": refresh_token,
+                }
+            )
+
+
+class OAuthServerUserAccountApi(Resource):
+    @setup_required
+    @oauth_server_client_id_required
+    @oauth_server_access_token_required
+    def post(self, oauth_provider_app: OAuthProviderApp, account: Account):
+        return jsonable_encoder(
+            {
+                "name": account.name,
+                "email": account.email,
+                "avatar": account.avatar,
+                "interface_language": account.interface_language,
+                "timezone": account.timezone,
+            }
+        )
+
+
+api.add_resource(OAuthServerAppApi, "/oauth/provider")
+api.add_resource(OAuthServerUserAuthorizeApi, "/oauth/provider/authorize")
+api.add_resource(OAuthServerUserTokenApi, "/oauth/provider/token")
+api.add_resource(OAuthServerUserAccountApi, "/oauth/provider/account")

api/controllers/console/billing/billing.py

@@ -1,9 +1,9 @@
-from flask_login import current_user
 from flask_restx import Resource, reqparse
 
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, only_edition_cloud, setup_required
-from libs.login import login_required
+from libs.login import current_user, login_required
+from models.model import Account
 from services.billing_service import BillingService
 
 
@@ -17,9 +17,10 @@ class Subscription(Resource):
         parser.add_argument("plan", type=str, required=True, location="args", choices=["professional", "team"])
         parser.add_argument("interval", type=str, required=True, location="args", choices=["month", "year"])
         args = parser.parse_args()
+        assert isinstance(current_user, Account)
 
         BillingService.is_tenant_owner_or_admin(current_user)
-
+        assert current_user.current_tenant_id is not None
         return BillingService.get_subscription(
             args["plan"], args["interval"], current_user.email, current_user.current_tenant_id
         )
@@ -31,7 +32,9 @@ class Invoices(Resource):
     @account_initialization_required
     @only_edition_cloud
     def get(self):
+        assert isinstance(current_user, Account)
         BillingService.is_tenant_owner_or_admin(current_user)
+        assert current_user.current_tenant_id is not None
         return BillingService.get_invoices(current_user.email, current_user.current_tenant_id)
 
 

api/controllers/console/datasets/data_source.py

@@ -10,6 +10,7 @@ from werkzeug.exceptions import NotFound
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, setup_required
 from core.indexing_runner import IndexingRunner
+from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.extractor.notion_extractor import NotionExtractor
 from extensions.ext_database import db
@@ -214,7 +215,7 @@ class DataSourceNotionApi(Resource):
             workspace_id = notion_info["workspace_id"]
             for page in notion_info["pages"]:
                 extract_setting = ExtractSetting(
-                    datasource_type="notion_import",
+                    datasource_type=DatasourceType.NOTION.value,
                     notion_info={
                         "notion_workspace_id": workspace_id,
                         "notion_obj_id": page["page_id"],

api/controllers/console/datasets/datasets.py

@@ -22,6 +22,7 @@ from core.model_runtime.entities.model_entities import ModelType
 from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
 from core.rag.datasource.vdb.vector_type import VectorType
+from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from extensions.ext_database import db
@@ -422,7 +423,9 @@ class DatasetIndexingEstimateApi(Resource):
             if file_details:
                 for file_detail in file_details:
                     extract_setting = ExtractSetting(
-                        datasource_type="upload_file", upload_file=file_detail, document_model=args["doc_form"]
+                        datasource_type=DatasourceType.FILE.value,
+                        upload_file=file_detail,
+                        document_model=args["doc_form"],
                     )
                     extract_settings.append(extract_setting)
         elif args["info_list"]["data_source_type"] == "notion_import":
@@ -431,7 +434,7 @@ class DatasetIndexingEstimateApi(Resource):
                 workspace_id = notion_info["workspace_id"]
                 for page in notion_info["pages"]:
                     extract_setting = ExtractSetting(
-                        datasource_type="notion_import",
+                        datasource_type=DatasourceType.NOTION.value,
                         notion_info={
                             "notion_workspace_id": workspace_id,
                             "notion_obj_id": page["page_id"],
@@ -445,7 +448,7 @@ class DatasetIndexingEstimateApi(Resource):
             website_info_list = args["info_list"]["website_info_list"]
             for url in website_info_list["urls"]:
                 extract_setting = ExtractSetting(
-                    datasource_type="website_crawl",
+                    datasource_type=DatasourceType.WEBSITE.value,
                     website_info={
                         "provider": website_info_list["provider"],
                         "job_id": website_info_list["job_id"],

api/controllers/console/datasets/datasets_document.py

@@ -40,6 +40,7 @@ from core.model_manager import ModelManager
 from core.model_runtime.entities.model_entities import ModelType
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.plugin.impl.exc import PluginDaemonClientSideError
+from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from extensions.ext_database import db
 from fields.document_fields import (
@@ -354,9 +355,6 @@ class DatasetInitApi(Resource):
         parser.add_argument("embedding_model_provider", type=str, required=False, nullable=True, location="json")
         args = parser.parse_args()
 
-        # The role of the current user in the ta table must be admin, owner, or editor, or dataset_operator
-        if not current_user.is_dataset_editor:
-            raise Forbidden()
         knowledge_config = KnowledgeConfig(**args)
         if knowledge_config.indexing_technique == "high_quality":
             if knowledge_config.embedding_model is None or knowledge_config.embedding_model_provider is None:
@@ -428,7 +426,7 @@ class DocumentIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type="upload_file", upload_file=file, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE.value, upload_file=file, document_model=document.doc_form
                 )
 
                 indexing_runner = IndexingRunner()
@@ -477,6 +475,8 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
             data_source_info = document.data_source_info_dict
 
             if document.data_source_type == "upload_file":
+                if not data_source_info:
+                    continue
                 file_id = data_source_info["upload_file_id"]
                 file_detail = (
                     db.session.query(UploadFile)
@@ -488,13 +488,15 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                     raise NotFound("File not found.")
 
                 extract_setting = ExtractSetting(
-                    datasource_type="upload_file", upload_file=file_detail, document_model=document.doc_form
+                    datasource_type=DatasourceType.FILE.value, upload_file=file_detail, document_model=document.doc_form
                 )
                 extract_settings.append(extract_setting)
 
             elif document.data_source_type == "notion_import":
+                if not data_source_info:
+                    continue
                 extract_setting = ExtractSetting(
-                    datasource_type="notion_import",
+                    datasource_type=DatasourceType.NOTION.value,
                     notion_info={
                         "notion_workspace_id": data_source_info["notion_workspace_id"],
                         "notion_obj_id": data_source_info["notion_page_id"],
@@ -505,8 +507,10 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 )
                 extract_settings.append(extract_setting)
             elif document.data_source_type == "website_crawl":
+                if not data_source_info:
+                    continue
                 extract_setting = ExtractSetting(
-                    datasource_type="website_crawl",
+                    datasource_type=DatasourceType.WEBSITE.value,
                     website_info={
                         "provider": data_source_info["provider"],
                         "job_id": data_source_info["job_id"],

api/controllers/console/explore/conversation.py

@@ -61,7 +61,6 @@ class ConversationApi(InstalledAppResource):
             ConversationService.delete(app_model, conversation_id, current_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
-        WebConversationService.unpin(app_model, conversation_id, current_user)
 
         return {"result": "success"}, 204
 

api/controllers/console/explore/parameter.py

@@ -43,6 +43,8 @@ class ExploreAppMetaApi(InstalledAppResource):
     def get(self, installed_app: InstalledApp):
         """Get app meta"""
         app_model = installed_app.app
+        if not app_model:
+            raise ValueError("App not found")
         return AppService().get_app_meta(app_model)
 
 

api/controllers/console/explore/workflow.py

@@ -35,6 +35,8 @@ class InstalledAppWorkflowRunApi(InstalledAppResource):
         Run workflow
         """
         app_model = installed_app.app
+        if not app_model:
+            raise NotWorkflowAppError()
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()
@@ -73,6 +75,8 @@ class InstalledAppWorkflowTaskStopApi(InstalledAppResource):
         Stop workflow task
         """
         app_model = installed_app.app
+        if not app_model:
+            raise NotWorkflowAppError()
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode != AppMode.WORKFLOW:
             raise NotWorkflowAppError()

api/controllers/console/explore/wraps.py

@@ -1,4 +1,6 @@
+from collections.abc import Callable
 from functools import wraps
+from typing import Concatenate, Optional, ParamSpec, TypeVar
 
 from flask_login import current_user
 from flask_restx import Resource
@@ -13,19 +15,15 @@ from services.app_service import AppService
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
+P = ParamSpec("P")
+R = TypeVar("R")
+T = TypeVar("T")
 
-def installed_app_required(view=None):
-    def decorator(view):
+
+def installed_app_required(view: Optional[Callable[Concatenate[InstalledApp, P], R]] = None):
+    def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
-            if not kwargs.get("installed_app_id"):
-                raise ValueError("missing installed_app_id in path parameters")
-
-            installed_app_id = kwargs.get("installed_app_id")
-            installed_app_id = str(installed_app_id)
-
-            del kwargs["installed_app_id"]
-
+        def decorated(installed_app_id: str, *args: P.args, **kwargs: P.kwargs):
             installed_app = (
                 db.session.query(InstalledApp)
                 .where(
@@ -52,10 +50,10 @@ def installed_app_required(view=None):
     return decorator
 
 
-def user_allowed_to_access_app(view=None):
-    def decorator(view):
+def user_allowed_to_access_app(view: Optional[Callable[Concatenate[InstalledApp, P], R]] = None):
+    def decorator(view: Callable[Concatenate[InstalledApp, P], R]):
         @wraps(view)
-        def decorated(installed_app: InstalledApp, *args, **kwargs):
+        def decorated(installed_app: InstalledApp, *args: P.args, **kwargs: P.kwargs):
             feature = FeatureService.get_system_features()
             if feature.webapp_auth.enabled:
                 app_id = installed_app.app_id

api/controllers/console/workspace/__init__.py

@@ -1,4 +1,6 @@
+from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask_login import current_user
 from sqlalchemy.orm import Session
@@ -7,14 +9,17 @@ from werkzeug.exceptions import Forbidden
 from extensions.ext_database import db
 from models.account import TenantPluginPermission
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 def plugin_permission_required(
     install_required: bool = False,
     debug_required: bool = False,
 ):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             user = current_user
             tenant_id = user.current_tenant_id
 

api/controllers/console/workspace/model_providers.py

@@ -67,7 +67,7 @@ class ModelProviderCredentialApi(Resource):
 
         parser = reqparse.RequestParser()
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         args = parser.parse_args()
 
         model_provider_service = ModelProviderService()
@@ -94,7 +94,7 @@ class ModelProviderCredentialApi(Resource):
         parser = reqparse.RequestParser()
         parser.add_argument("credential_id", type=uuid_value, required=True, nullable=False, location="json")
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         args = parser.parse_args()
 
         model_provider_service = ModelProviderService()

api/controllers/console/workspace/models.py

@@ -219,7 +219,11 @@ class ModelProviderModelCredentialApi(Resource):
 
         model_load_balancing_service = ModelLoadBalancingService()
         is_load_balancing_enabled, load_balancing_configs = model_load_balancing_service.get_load_balancing_configs(
-            tenant_id=tenant_id, provider=provider, model=args["model"], model_type=args["model_type"]
+            tenant_id=tenant_id,
+            provider=provider,
+            model=args["model"],
+            model_type=args["model_type"],
+            config_from=args.get("config_from", ""),
         )
 
         if args.get("config_from", "") == "predefined-model":
@@ -263,7 +267,7 @@ class ModelProviderModelCredentialApi(Resource):
             choices=[mt.value for mt in ModelType],
             location="json",
         )
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
         args = parser.parse_args()
 
@@ -309,7 +313,7 @@ class ModelProviderModelCredentialApi(Resource):
         )
         parser.add_argument("credential_id", type=uuid_value, required=True, nullable=False, location="json")
         parser.add_argument("credentials", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("name", type=StrLen(30), required=True, nullable=False, location="json")
+        parser.add_argument("name", type=StrLen(30), required=False, nullable=True, location="json")
         args = parser.parse_args()
 
         model_provider_service = ModelProviderService()

api/controllers/console/wraps.py

@@ -2,7 +2,9 @@ import contextlib
 import json
 import os
 import time
+from collections.abc import Callable
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import abort, request
 from flask_login import current_user
@@ -19,10 +21,13 @@ from services.operation_service import OperationService
 
 from .error import NotInitValidateError, NotSetupError, UnauthorizedAndForceLogout
 
+P = ParamSpec("P")
+R = TypeVar("R")
 
-def account_initialization_required(view):
+
+def account_initialization_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check account initialization
         account = current_user
 
@@ -34,9 +39,9 @@ def account_initialization_required(view):
     return decorated
 
 
-def only_edition_cloud(view):
+def only_edition_cloud(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if dify_config.EDITION != "CLOUD":
             abort(404)
 
@@ -45,9 +50,9 @@ def only_edition_cloud(view):
     return decorated
 
 
-def only_edition_enterprise(view):
+def only_edition_enterprise(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.ENTERPRISE_ENABLED:
             abort(404)
 
@@ -56,9 +61,9 @@ def only_edition_enterprise(view):
     return decorated
 
 
-def only_edition_self_hosted(view):
+def only_edition_self_hosted(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if dify_config.EDITION != "SELF_HOSTED":
             abort(404)
 
@@ -67,9 +72,9 @@ def only_edition_self_hosted(view):
     return decorated
 
 
-def cloud_edition_billing_enabled(view):
+def cloud_edition_billing_enabled(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_features(current_user.current_tenant_id)
         if not features.billing.enabled:
             abort(403, "Billing feature is not enabled.")
@@ -79,9 +84,9 @@ def cloud_edition_billing_enabled(view):
 
 
 def cloud_edition_billing_resource_check(resource: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             features = FeatureService.get_features(current_user.current_tenant_id)
             if features.billing.enabled:
                 members = features.members
@@ -120,9 +125,9 @@ def cloud_edition_billing_resource_check(resource: str):
 
 
 def cloud_edition_billing_knowledge_limit_check(resource: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             features = FeatureService.get_features(current_user.current_tenant_id)
             if features.billing.enabled:
                 if resource == "add_segment":
@@ -142,9 +147,9 @@ def cloud_edition_billing_knowledge_limit_check(resource: str):
 
 
 def cloud_edition_billing_rate_limit_check(resource: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             if resource == "knowledge":
                 knowledge_rate_limit = FeatureService.get_knowledge_rate_limit(current_user.current_tenant_id)
                 if knowledge_rate_limit.enabled:
@@ -176,9 +181,9 @@ def cloud_edition_billing_rate_limit_check(resource: str):
     return interceptor
 
 
-def cloud_utm_record(view):
+def cloud_utm_record(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         with contextlib.suppress(Exception):
             features = FeatureService.get_features(current_user.current_tenant_id)
 
@@ -194,9 +199,9 @@ def cloud_utm_record(view):
     return decorated
 
 
-def setup_required(view):
+def setup_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         # check setup
         if (
             dify_config.EDITION == "SELF_HOSTED"
@@ -212,9 +217,9 @@ def setup_required(view):
     return decorated
 
 
-def enterprise_license_required(view):
+def enterprise_license_required(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         settings = FeatureService.get_system_features()
         if settings.license.status in [LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST]:
             raise UnauthorizedAndForceLogout("Your license is invalid. Please contact your administrator.")
@@ -224,9 +229,9 @@ def enterprise_license_required(view):
     return decorated
 
 
-def email_password_login_enabled(view):
+def email_password_login_enabled(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if features.enable_email_password_login:
             return view(*args, **kwargs)
@@ -237,9 +242,9 @@ def email_password_login_enabled(view):
     return decorated
 
 
-def enable_change_email(view):
+def enable_change_email(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_system_features()
         if features.enable_change_email:
             return view(*args, **kwargs)
@@ -250,9 +255,9 @@ def enable_change_email(view):
     return decorated
 
 
-def is_allow_transfer_owner(view):
+def is_allow_transfer_owner(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         features = FeatureService.get_features(current_user.current_tenant_id)
         if features.is_allow_transfer_workspace:
             return view(*args, **kwargs)

api/controllers/inner_api/wraps.py

@@ -1,8 +1,12 @@
 from base64 import b64encode
+from collections.abc import Callable
 from functools import wraps
 from hashlib import sha1
 from hmac import new as hmac_new
+from typing import ParamSpec, TypeVar
 
+P = ParamSpec("P")
+R = TypeVar("R")
 from flask import abort, request
 
 from configs import dify_config
@@ -10,9 +14,9 @@ from extensions.ext_database import db
 from models.model import EndUser
 
 
-def billing_inner_api_only(view):
+def billing_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.INNER_API:
             abort(404)
 
@@ -26,9 +30,9 @@ def billing_inner_api_only(view):
     return decorated
 
 
-def enterprise_inner_api_only(view):
+def enterprise_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.INNER_API:
             abort(404)
 
@@ -78,9 +82,9 @@ def enterprise_inner_api_user_auth(view):
     return decorated
 
 
-def plugin_inner_api_only(view):
+def plugin_inner_api_only(view: Callable[P, R]):
     @wraps(view)
-    def decorated(*args, **kwargs):
+    def decorated(*args: P.args, **kwargs: P.kwargs):
         if not dify_config.PLUGIN_DAEMON_KEY:
             abort(404)
 

api/controllers/mcp/mcp.py

@@ -1,18 +1,27 @@
 from typing import Optional, Union
 
+from flask import Response
 from flask_restx import Resource, reqparse
 from pydantic import ValidationError
+from sqlalchemy.orm import Session
 
 from controllers.console.app.mcp_server import AppMCPServerStatus
 from controllers.mcp import mcp_ns
 from core.app.app_config.entities import VariableEntity
-from core.mcp import types
-from core.mcp.server.streamable_http import MCPServerStreamableHTTPRequestHandler
-from core.mcp.types import ClientNotification, ClientRequest
-from core.mcp.utils import create_mcp_error_response
+from core.mcp import types as mcp_types
+from core.mcp.server.streamable_http import handle_mcp_request
 from extensions.ext_database import db
 from libs import helper
-from models.model import App, AppMCPServer, AppMode
+from models.model import App, AppMCPServer, AppMode, EndUser
+
+
+class MCPRequestError(Exception):
+    """Custom exception for MCP request processing errors"""
+
+    def __init__(self, error_code: int, message: str):
+        self.error_code = error_code
+        self.message = message
+        super().__init__(message)
 
 
 def int_or_str(value):
@@ -63,77 +72,173 @@ class MCPAppApi(Resource):
         Raises:
             ValidationError: Invalid request format or parameters
         """
-        # Parse and validate all arguments
         args = mcp_request_parser.parse_args()
-
         request_id: Optional[Union[int, str]] = args.get("id")
+        mcp_request = self._parse_mcp_request(args)
 
-        server = db.session.query(AppMCPServer).where(AppMCPServer.server_code == server_code).first()
-        if not server:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_REQUEST, "Server Not Found")
-            )
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get MCP server and app
+            mcp_server, app = self._get_mcp_server_and_app(server_code, session)
+            self._validate_server_status(mcp_server)
 
-        if server.status != AppMCPServerStatus.ACTIVE:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_REQUEST, "Server is not active")
-            )
+            # Get user input form
+            user_input_form = self._get_user_input_form(app)
 
-        app = db.session.query(App).where(App.id == server.app_id).first()
+            # Handle notification vs request differently
+            return self._process_mcp_message(mcp_request, request_id, app, mcp_server, user_input_form, session)
+
+    def _get_mcp_server_and_app(self, server_code: str, session: Session) -> tuple[AppMCPServer, App]:
+        """Get and validate MCP server and app in one query session"""
+        mcp_server = session.query(AppMCPServer).where(AppMCPServer.server_code == server_code).first()
+        if not mcp_server:
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Server Not Found")
+
+        app = session.query(App).where(App.id == mcp_server.app_id).first()
         if not app:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_REQUEST, "App Not Found")
-            )
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "App Not Found")
 
-        if app.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
-            workflow = app.workflow
-            if workflow is None:
-                return helper.compact_generate_response(
-                    create_mcp_error_response(request_id, types.INVALID_REQUEST, "App is unavailable")
-                )
+        return mcp_server, app
 
-            user_input_form = workflow.user_input_form(to_old_structure=True)
+    def _validate_server_status(self, mcp_server: AppMCPServer):
+        """Validate MCP server status"""
+        if mcp_server.status != AppMCPServerStatus.ACTIVE:
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Server is not active")
+
+    def _process_mcp_message(
+        self,
+        mcp_request: mcp_types.ClientRequest | mcp_types.ClientNotification,
+        request_id: Optional[Union[int, str]],
+        app: App,
+        mcp_server: AppMCPServer,
+        user_input_form: list[VariableEntity],
+        session: Session,
+    ) -> Response:
+        """Process MCP message (notification or request)"""
+        if isinstance(mcp_request, mcp_types.ClientNotification):
+            return self._handle_notification(mcp_request)
         else:
-            app_model_config = app.app_model_config
-            if app_model_config is None:
-                return helper.compact_generate_response(
-                    create_mcp_error_response(request_id, types.INVALID_REQUEST, "App is unavailable")
-                )
+            return self._handle_request(mcp_request, request_id, app, mcp_server, user_input_form, session)
 
-            features_dict = app_model_config.to_dict()
-            user_input_form = features_dict.get("user_input_form", [])
-        converted_user_input_form: list[VariableEntity] = []
-        try:
-            for item in user_input_form:
-                variable_type = item.get("type", "") or list(item.keys())[0]
-                variable = item[variable_type]
-                converted_user_input_form.append(
-                    VariableEntity(
-                        type=variable_type,
-                        variable=variable.get("variable"),
-                        description=variable.get("description") or "",
-                        label=variable.get("label"),
-                        required=variable.get("required", False),
-                        max_length=variable.get("max_length"),
-                        options=variable.get("options") or [],
-                    )
-                )
-        except ValidationError as e:
-            return helper.compact_generate_response(
-                create_mcp_error_response(request_id, types.INVALID_PARAMS, f"Invalid user_input_form: {str(e)}")
-            )
+    def _handle_notification(self, mcp_request: mcp_types.ClientNotification) -> Response:
+        """Handle MCP notification"""
+        # For notifications, only support init notification
+        if mcp_request.root.method != "notifications/initialized":
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Invalid notification method")
+        # Return HTTP 202 Accepted for notifications (no response body)
+        return Response("", status=202, content_type="application/json")
 
+    def _handle_request(
+        self,
+        mcp_request: mcp_types.ClientRequest,
+        request_id: Optional[Union[int, str]],
+        app: App,
+        mcp_server: AppMCPServer,
+        user_input_form: list[VariableEntity],
+        session: Session,
+    ) -> Response:
+        """Handle MCP request"""
+        if request_id is None:
+            raise MCPRequestError(mcp_types.INVALID_REQUEST, "Request ID is required")
+
+        result = self._handle_mcp_request(app, mcp_server, mcp_request, user_input_form, session, request_id)
+        if result is None:
+            # This shouldn't happen for requests, but handle gracefully
+            raise MCPRequestError(mcp_types.INTERNAL_ERROR, "No response generated for request")
+
+        return helper.compact_generate_response(result.model_dump(by_alias=True, mode="json", exclude_none=True))
+
+    def _get_user_input_form(self, app: App) -> list[VariableEntity]:
+        """Get and convert user input form"""
+        # Get raw user input form based on app mode
+        if app.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
+            if not app.workflow:
+                raise MCPRequestError(mcp_types.INVALID_REQUEST, "App is unavailable")
+            raw_user_input_form = app.workflow.user_input_form(to_old_structure=True)
+        else:
+            if not app.app_model_config:
+                raise MCPRequestError(mcp_types.INVALID_REQUEST, "App is unavailable")
+            features_dict = app.app_model_config.to_dict()
+            raw_user_input_form = features_dict.get("user_input_form", [])
+
+        # Convert to VariableEntity objects
         try:
-            request: ClientRequest | ClientNotification = ClientRequest.model_validate(args)
+            return self._convert_user_input_form(raw_user_input_form)
         except ValidationError as e:
+            raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid user_input_form: {str(e)}")
+
+    def _convert_user_input_form(self, raw_form: list[dict]) -> list[VariableEntity]:
+        """Convert raw user input form to VariableEntity objects"""
+        return [self._create_variable_entity(item) for item in raw_form]
+
+    def _create_variable_entity(self, item: dict) -> VariableEntity:
+        """Create a single VariableEntity from raw form item"""
+        variable_type = item.get("type", "") or list(item.keys())[0]
+        variable = item[variable_type]
+
+        return VariableEntity(
+            type=variable_type,
+            variable=variable.get("variable"),
+            description=variable.get("description") or "",
+            label=variable.get("label"),
+            required=variable.get("required", False),
+            max_length=variable.get("max_length"),
+            options=variable.get("options") or [],
+        )
+
+    def _parse_mcp_request(self, args: dict) -> mcp_types.ClientRequest | mcp_types.ClientNotification:
+        """Parse and validate MCP request"""
+        try:
+            return mcp_types.ClientRequest.model_validate(args)
+        except ValidationError:
             try:
-                notification = ClientNotification.model_validate(args)
-                request = notification
+                return mcp_types.ClientNotification.model_validate(args)
             except ValidationError as e:
-                return helper.compact_generate_response(
-                    create_mcp_error_response(request_id, types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
-                )
+                raise MCPRequestError(mcp_types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
 
-        mcp_server_handler = MCPServerStreamableHTTPRequestHandler(app, request, converted_user_input_form)
-        response = mcp_server_handler.handle()
-        return helper.compact_generate_response(response)
+    def _retrieve_end_user(self, tenant_id: str, mcp_server_id: str, session: Session) -> EndUser | None:
+        """Get end user from existing session - optimized query"""
+        return (
+            session.query(EndUser)
+            .where(EndUser.tenant_id == tenant_id)
+            .where(EndUser.session_id == mcp_server_id)
+            .where(EndUser.type == "mcp")
+            .first()
+        )
+
+    def _create_end_user(
+        self, client_name: str, tenant_id: str, app_id: str, mcp_server_id: str, session: Session
+    ) -> EndUser:
+        """Create end user in existing session"""
+        end_user = EndUser(
+            tenant_id=tenant_id,
+            app_id=app_id,
+            type="mcp",
+            name=client_name,
+            session_id=mcp_server_id,
+        )
+        session.add(end_user)
+        session.flush()  # Use flush instead of commit to keep transaction open
+        session.refresh(end_user)
+        return end_user
+
+    def _handle_mcp_request(
+        self,
+        app: App,
+        mcp_server: AppMCPServer,
+        mcp_request: mcp_types.ClientRequest,
+        user_input_form: list[VariableEntity],
+        session: Session,
+        request_id: Union[int, str],
+    ) -> mcp_types.JSONRPCResponse | mcp_types.JSONRPCError | None:
+        """Handle MCP request and return response"""
+        end_user = self._retrieve_end_user(mcp_server.tenant_id, mcp_server.id, session)
+
+        if not end_user and isinstance(mcp_request.root, mcp_types.InitializeRequest):
+            client_info = mcp_request.root.params.clientInfo
+            client_name = f"{client_info.name}@{client_info.version}"
+            # Commit the session before creating end user to avoid transaction conflicts
+            session.commit()
+            with Session(db.engine, expire_on_commit=False) as create_session, create_session.begin():
+                end_user = self._create_end_user(client_name, app.tenant_id, app.id, mcp_server.id, create_session)
+
+        return handle_mcp_request(app, mcp_request, user_input_form, mcp_server, end_user, request_id)

api/controllers/service_api/app/audio.py

@@ -55,7 +55,7 @@ class AudioApi(Resource):
         file = request.files["file"]
 
         try:
-            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user)
+            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=end_user.id)
 
             return response
         except services.errors.app_model_config.AppModelConfigBrokenError:

api/controllers/service_api/app/file_preview.py

@@ -59,7 +59,7 @@ class FilePreviewApi(Resource):
         args = file_preview_parser.parse_args()
 
         # Validate file ownership and get file objects
-        message_file, upload_file = self._validate_file_ownership(file_id, app_model.id)
+        _, upload_file = self._validate_file_ownership(file_id, app_model.id)
 
         # Get file content generator
         try:

api/controllers/service_api/dataset/dataset.py

@@ -318,10 +318,6 @@ class DatasetApi(DatasetApiResource):
         except services.errors.account.NoPermissionError as e:
             raise Forbidden(str(e))
         data = marshal(dataset, dataset_detail_fields)
-        if data.get("permission") == "partial_members":
-            part_users_list = DatasetPermissionService.get_dataset_partial_member_list(dataset_id_str)
-            data.update({"partial_member_list": part_users_list})
-
         # check embedding setting
         provider_manager = ProviderManager()
         assert isinstance(current_user, Account)

api/controllers/service_api/dataset/document.py

@@ -410,7 +410,7 @@ class DocumentUpdateByFileApi(DatasetApiResource):
         DocumentService.document_create_args_validate(knowledge_config)
 
         try:
-            documents, batch = DocumentService.save_document_with_dataset_id(
+            documents, _ = DocumentService.save_document_with_dataset_id(
                 dataset=dataset,
                 knowledge_config=knowledge_config,
                 account=dataset.created_by_account,

api/controllers/service_api/dataset/metadata.py

@@ -1,6 +1,6 @@
 from typing import Literal
 
-from flask_login import current_user  # type: ignore
+from flask_login import current_user
 from flask_restx import marshal, reqparse
 from werkzeug.exceptions import NotFound
 

api/controllers/service_api/dataset/segment.py

@@ -440,7 +440,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Segment not found.")
 
         # validate segment belongs to the specified document
-        if segment.document_id != document_id:
+        if str(segment.document_id) != str(document_id):
             raise NotFound("Document not found.")
 
         # check child chunk
@@ -451,7 +451,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Child chunk not found.")
 
         # validate child chunk belongs to the specified segment
-        if child_chunk.segment_id != segment.id:
+        if str(child_chunk.segment_id) != str(segment.id):
             raise NotFound("Child chunk not found.")
 
         try:
@@ -500,7 +500,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Segment not found.")
 
         # validate segment belongs to the specified document
-        if segment.document_id != document_id:
+        if str(segment.document_id) != str(document_id):
             raise NotFound("Segment not found.")
 
         # get child chunk
@@ -511,7 +511,7 @@ class DatasetChildChunkApi(DatasetApiResource):
             raise NotFound("Child chunk not found.")
 
         # validate child chunk belongs to the specified segment
-        if child_chunk.segment_id != segment.id:
+        if str(child_chunk.segment_id) != str(segment.id):
             raise NotFound("Child chunk not found.")
 
         # validate args

api/controllers/service_api/wraps.py

@@ -1,12 +1,12 @@
 import time
 from collections.abc import Callable
 from datetime import timedelta
-from enum import Enum
+from enum import StrEnum, auto
 from functools import wraps
-from typing import Optional
+from typing import Optional, ParamSpec, TypeVar
 
 from flask import current_app, request
-from flask_login import user_logged_in  # type: ignore
+from flask_login import user_logged_in
 from flask_restx import Resource
 from pydantic import BaseModel
 from sqlalchemy import select, update
@@ -22,15 +22,18 @@ from models.dataset import Dataset, RateLimitLog
 from models.model import ApiToken, App, EndUser
 from services.feature_service import FeatureService
 
+P = ParamSpec("P")
+R = TypeVar("R")
 
-class WhereisUserArg(Enum):
+
+class WhereisUserArg(StrEnum):
     """
     Enum for whereis_user_arg.
     """
 
-    QUERY = "query"
-    JSON = "json"
-    FORM = "form"
+    QUERY = auto()
+    JSON = auto()
+    FORM = auto()
 
 
 class FetchUserArg(BaseModel):
@@ -60,27 +63,6 @@ def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optio
             if tenant.status == TenantStatus.ARCHIVE:
                 raise Forbidden("The workspace's status is archived.")
 
-            tenant_account_join = (
-                db.session.query(Tenant, TenantAccountJoin)
-                .where(Tenant.id == api_token.tenant_id)
-                .where(TenantAccountJoin.tenant_id == Tenant.id)
-                .where(TenantAccountJoin.role.in_(["owner"]))
-                .where(Tenant.status == TenantStatus.NORMAL)
-                .one_or_none()
-            )  # TODO: only owner information is required, so only one is returned.
-            if tenant_account_join:
-                tenant, ta = tenant_account_join
-                account = db.session.query(Account).where(Account.id == ta.account_id).first()
-                # Login admin
-                if account:
-                    account.current_tenant = tenant
-                    current_app.login_manager._update_request_context_with_user(account)  # type: ignore
-                    user_logged_in.send(current_app._get_current_object(), user=_get_user())  # type: ignore
-                else:
-                    raise Unauthorized("Tenant owner account does not exist.")
-            else:
-                raise Unauthorized("Tenant does not exist.")
-
             kwargs["app_model"] = app_model
 
             if fetch_user_arg:
@@ -118,8 +100,8 @@ def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optio
 
 
 def cloud_edition_billing_resource_check(resource: str, api_token_type: str):
-    def interceptor(view):
-        def decorated(*args, **kwargs):
+    def interceptor(view: Callable[P, R]):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
             features = FeatureService.get_features(api_token.tenant_id)
 
@@ -148,9 +130,9 @@ def cloud_edition_billing_resource_check(resource: str, api_token_type: str):
 
 
 def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
             features = FeatureService.get_features(api_token.tenant_id)
             if features.billing.enabled:
@@ -170,9 +152,9 @@ def cloud_edition_billing_knowledge_limit_check(resource: str, api_token_type: s
 
 
 def cloud_edition_billing_rate_limit_check(resource: str, api_token_type: str):
-    def interceptor(view):
+    def interceptor(view: Callable[P, R]):
         @wraps(view)
-        def decorated(*args, **kwargs):
+        def decorated(*args: P.args, **kwargs: P.kwargs):
             api_token = validate_and_get_api_token(api_token_type)
 
             if resource == "knowledge":
@@ -291,27 +273,28 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: Optional[str]
     if not user_id:
         user_id = "DEFAULT-USER"
 
-    end_user = (
-        db.session.query(EndUser)
-        .where(
-            EndUser.tenant_id == app_model.tenant_id,
-            EndUser.app_id == app_model.id,
-            EndUser.session_id == user_id,
-            EndUser.type == "service_api",
+    with Session(db.engine, expire_on_commit=False) as session:
+        end_user = (
+            session.query(EndUser)
+            .where(
+                EndUser.tenant_id == app_model.tenant_id,
+                EndUser.app_id == app_model.id,
+                EndUser.session_id == user_id,
+                EndUser.type == "service_api",
+            )
+            .first()
         )
-        .first()
-    )
 
-    if end_user is None:
-        end_user = EndUser(
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            type="service_api",
-            is_anonymous=user_id == "DEFAULT-USER",
-            session_id=user_id,
-        )
-        db.session.add(end_user)
-        db.session.commit()
+        if end_user is None:
+            end_user = EndUser(
+                tenant_id=app_model.tenant_id,
+                app_id=app_model.id,
+                type="service_api",
+                is_anonymous=user_id == "DEFAULT-USER",
+                session_id=user_id,
+            )
+            session.add(end_user)
+            session.commit()
 
     return end_user
 

api/controllers/web/__init__.py

@@ -1,19 +1,20 @@
 from flask import Blueprint
+from flask_restx import Namespace
 
 from libs.external_api import ExternalApi
 
-from .files import FileApi
-from .remote_files import RemoteFileInfoApi, RemoteFileUploadApi
-
 bp = Blueprint("web", __name__, url_prefix="/api")
-api = ExternalApi(bp)
 
-# Files
-api.add_resource(FileApi, "/files/upload")
+api = ExternalApi(
+    bp,
+    version="1.0",
+    title="Web API",
+    description="Public APIs for web applications including file uploads, chat interactions, and app management",
+    doc="/docs",  # Enable Swagger UI at /api/docs
+)
 
-# Remote files
-api.add_resource(RemoteFileInfoApi, "/remote-files/<path:url>")
-api.add_resource(RemoteFileUploadApi, "/remote-files/upload")
+# Create namespace
+web_ns = Namespace("web", description="Web application API operations", path="/")
 
 from . import (
     app,
@@ -21,11 +22,15 @@ from . import (
     completion,
     conversation,
     feature,
+    files,
     forgot_password,
     login,
     message,
     passport,
+    remote_files,
     saved_message,
     site,
     workflow,
 )
+
+api.add_namespace(web_ns)

api/controllers/web/app.py

@@ -5,7 +5,7 @@ from flask_restx import Resource, marshal_with, reqparse
 from werkzeug.exceptions import Unauthorized
 
 from controllers.common import fields
-from controllers.web import api
+from controllers.web import web_ns
 from controllers.web.error import AppUnavailableError
 from controllers.web.wraps import WebApiResource
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
@@ -19,9 +19,22 @@ from services.webapp_auth_service import WebAppAuthService
 logger = logging.getLogger(__name__)
 
 
+@web_ns.route("/parameters")
 class AppParameterApi(WebApiResource):
     """Resource for app variables."""
 
+    @web_ns.doc("Get App Parameters")
+    @web_ns.doc(description="Retrieve the parameters for a specific app.")
+    @web_ns.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "App Not Found",
+            500: "Internal Server Error",
+        }
+    )
     @marshal_with(fields.parameters_fields)
     def get(self, app_model: App, end_user):
         """Retrieve app parameters."""
@@ -44,13 +57,42 @@ class AppParameterApi(WebApiResource):
         return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
 
 
+@web_ns.route("/meta")
 class AppMeta(WebApiResource):
+    @web_ns.doc("Get App Meta")
+    @web_ns.doc(description="Retrieve the metadata for a specific app.")
+    @web_ns.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "App Not Found",
+            500: "Internal Server Error",
+        }
+    )
     def get(self, app_model: App, end_user):
         """Get app meta"""
         return AppService().get_app_meta(app_model)
 
 
+@web_ns.route("/webapp/access-mode")
 class AppAccessMode(Resource):
+    @web_ns.doc("Get App Access Mode")
+    @web_ns.doc(description="Retrieve the access mode for a web application (public or restricted).")
+    @web_ns.doc(
+        params={
+            "appId": {"description": "Application ID", "type": "string", "required": False},
+            "appCode": {"description": "Application code", "type": "string", "required": False},
+        }
+    )
+    @web_ns.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            500: "Internal Server Error",
+        }
+    )
     def get(self):
         parser = reqparse.RequestParser()
         parser.add_argument("appId", type=str, required=False, location="args")
@@ -74,7 +116,19 @@ class AppAccessMode(Resource):
         return {"accessMode": res.access_mode}
 
 
+@web_ns.route("/webapp/permission")
 class AppWebAuthPermission(Resource):
+    @web_ns.doc("Check App Permission")
+    @web_ns.doc(description="Check if user has permission to access a web application.")
+    @web_ns.doc(params={"appId": {"description": "Application ID", "type": "string", "required": True}})
+    @web_ns.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            500: "Internal Server Error",
+        }
+    )
     def get(self):
         user_id = "visitor"
         try:
@@ -112,10 +166,3 @@ class AppWebAuthPermission(Resource):
         if WebAppAuthService.is_app_require_permission_check(app_id=app_id):
             res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(str(user_id), app_code)
         return {"result": res}
-
-
-api.add_resource(AppParameterApi, "/parameters")
-api.add_resource(AppMeta, "/meta")
-# webapp auth apis
-api.add_resource(AppAccessMode, "/webapp/access-mode")
-api.add_resource(AppWebAuthPermission, "/webapp/permission")

api/controllers/web/audio.py

@@ -1,6 +1,7 @@
 import logging
 
 from flask import request
+from flask_restx import fields, marshal_with, reqparse
 from werkzeug.exceptions import InternalServerError
 
 import services
@@ -32,7 +33,26 @@ logger = logging.getLogger(__name__)
 
 
 class AudioApi(WebApiResource):
+    audio_to_text_response_fields = {
+        "text": fields.String,
+    }
+
+    @marshal_with(audio_to_text_response_fields)
+    @api.doc("Audio to Text")
+    @api.doc(description="Convert audio file to text using speech-to-text service.")
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            413: "Audio file too large",
+            415: "Unsupported audio type",
+            500: "Internal Server Error",
+        }
+    )
     def post(self, app_model: App, end_user):
+        """Convert audio to text"""
         file = request.files["file"]
 
         try:
@@ -66,9 +86,25 @@ class AudioApi(WebApiResource):
 
 
 class TextApi(WebApiResource):
-    def post(self, app_model: App, end_user):
-        from flask_restx import reqparse
+    text_to_audio_response_fields = {
+        "audio_url": fields.String,
+        "duration": fields.Float,
+    }
 
+    @marshal_with(text_to_audio_response_fields)
+    @api.doc("Text to Audio")
+    @api.doc(description="Convert text to audio using text-to-speech service.")
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            500: "Internal Server Error",
+        }
+    )
+    def post(self, app_model: App, end_user):
+        """Convert text to audio"""
         try:
             parser = reqparse.RequestParser()
             parser.add_argument("message_id", type=str, required=False, location="json")

api/controllers/web/completion.py

@@ -36,6 +36,32 @@ logger = logging.getLogger(__name__)
 
 # define completion api for user
 class CompletionApi(WebApiResource):
+    @api.doc("Create Completion Message")
+    @api.doc(description="Create a completion message for text generation applications.")
+    @api.doc(
+        params={
+            "inputs": {"description": "Input variables for the completion", "type": "object", "required": True},
+            "query": {"description": "Query text for completion", "type": "string", "required": False},
+            "files": {"description": "Files to be processed", "type": "array", "required": False},
+            "response_mode": {
+                "description": "Response mode: blocking or streaming",
+                "type": "string",
+                "enum": ["blocking", "streaming"],
+                "required": False,
+            },
+            "retriever_from": {"description": "Source of retriever", "type": "string", "required": False},
+        }
+    )
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "App Not Found",
+            500: "Internal Server Error",
+        }
+    )
     def post(self, app_model, end_user):
         if app_model.mode != "completion":
             raise NotCompletionAppError()
@@ -81,6 +107,19 @@ class CompletionApi(WebApiResource):
 
 
 class CompletionStopApi(WebApiResource):
+    @api.doc("Stop Completion Message")
+    @api.doc(description="Stop a running completion message task.")
+    @api.doc(params={"task_id": {"description": "Task ID to stop", "type": "string", "required": True}})
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "Task Not Found",
+            500: "Internal Server Error",
+        }
+    )
     def post(self, app_model, end_user, task_id):
         if app_model.mode != "completion":
             raise NotCompletionAppError()
@@ -91,6 +130,34 @@ class CompletionStopApi(WebApiResource):
 
 
 class ChatApi(WebApiResource):
+    @api.doc("Create Chat Message")
+    @api.doc(description="Create a chat message for conversational applications.")
+    @api.doc(
+        params={
+            "inputs": {"description": "Input variables for the chat", "type": "object", "required": True},
+            "query": {"description": "User query/message", "type": "string", "required": True},
+            "files": {"description": "Files to be processed", "type": "array", "required": False},
+            "response_mode": {
+                "description": "Response mode: blocking or streaming",
+                "type": "string",
+                "enum": ["blocking", "streaming"],
+                "required": False,
+            },
+            "conversation_id": {"description": "Conversation UUID", "type": "string", "required": False},
+            "parent_message_id": {"description": "Parent message UUID", "type": "string", "required": False},
+            "retriever_from": {"description": "Source of retriever", "type": "string", "required": False},
+        }
+    )
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "App Not Found",
+            500: "Internal Server Error",
+        }
+    )
     def post(self, app_model, end_user):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
@@ -141,6 +208,19 @@ class ChatApi(WebApiResource):
 
 
 class ChatStopApi(WebApiResource):
+    @api.doc("Stop Chat Message")
+    @api.doc(description="Stop a running chat message task.")
+    @api.doc(params={"task_id": {"description": "Task ID to stop", "type": "string", "required": True}})
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "Task Not Found",
+            500: "Internal Server Error",
+        }
+    )
     def post(self, app_model, end_user, task_id):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:

api/controllers/web/conversation.py

@@ -1,4 +1,4 @@
-from flask_restx import marshal_with, reqparse
+from flask_restx import fields, marshal_with, reqparse
 from flask_restx.inputs import int_range
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import NotFound
@@ -58,6 +58,11 @@ class ConversationListApi(WebApiResource):
 
 
 class ConversationApi(WebApiResource):
+    delete_response_fields = {
+        "result": fields.String,
+    }
+
+    @marshal_with(delete_response_fields)
     def delete(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
@@ -68,8 +73,6 @@ class ConversationApi(WebApiResource):
             ConversationService.delete(app_model, conversation_id, end_user)
         except ConversationNotExistsError:
             raise NotFound("Conversation Not Exists.")
-        WebConversationService.unpin(app_model, conversation_id, end_user)
-
         return {"result": "success"}, 204
 
 
@@ -94,6 +97,11 @@ class ConversationRenameApi(WebApiResource):
 
 
 class ConversationPinApi(WebApiResource):
+    pin_response_fields = {
+        "result": fields.String,
+    }
+
+    @marshal_with(pin_response_fields)
     def patch(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
@@ -110,6 +118,11 @@ class ConversationPinApi(WebApiResource):
 
 
 class ConversationUnPinApi(WebApiResource):
+    unpin_response_fields = {
+        "result": fields.String,
+    }
+
+    @marshal_with(unpin_response_fields)
     def patch(self, app_model, end_user, c_id):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:

api/controllers/web/feature.py

@@ -1,12 +1,21 @@
 from flask_restx import Resource
 
-from controllers.web import api
+from controllers.web import web_ns
 from services.feature_service import FeatureService
 
 
+@web_ns.route("/system-features")
 class SystemFeatureApi(Resource):
+    @web_ns.doc("get_system_features")
+    @web_ns.doc(description="Get system feature flags and configuration")
+    @web_ns.doc(responses={200: "System features retrieved successfully", 500: "Internal server error"})
     def get(self):
+        """Get system feature flags and configuration.
+
+        Returns the current system feature flags and configuration
+        that control various functionalities across the platform.
+
+        Returns:
+            dict: System feature configuration object
+        """
         return FeatureService.get_system_features().model_dump()
-
-
-api.add_resource(SystemFeatureApi, "/system-features")

api/controllers/web/files.py

@@ -9,14 +9,50 @@ from controllers.common.errors import (
     TooManyFilesError,
     UnsupportedFileTypeError,
 )
+from controllers.web import web_ns
 from controllers.web.wraps import WebApiResource
-from fields.file_fields import file_fields
+from fields.file_fields import build_file_model
 from services.file_service import FileService
 
 
+@web_ns.route("/files/upload")
 class FileApi(WebApiResource):
-    @marshal_with(file_fields)
+    @web_ns.doc("upload_file")
+    @web_ns.doc(description="Upload a file for use in web applications")
+    @web_ns.doc(
+        responses={
+            201: "File uploaded successfully",
+            400: "Bad request - invalid file or parameters",
+            413: "File too large",
+            415: "Unsupported file type",
+        }
+    )
+    @marshal_with(build_file_model(web_ns))
     def post(self, app_model, end_user):
+        """Upload a file for use in web applications.
+
+        Accepts file uploads for use within web applications, supporting
+        multiple file types with automatic validation and storage.
+
+        Args:
+            app_model: The associated application model
+            end_user: The end user uploading the file
+
+        Form Parameters:
+            file: The file to upload (required)
+            source: Optional source type (datasets or None)
+
+        Returns:
+            dict: File information including ID, URL, and metadata
+            int: HTTP status code 201 for success
+
+        Raises:
+            NoFileUploadedError: No file provided in request
+            TooManyFilesError: Multiple files provided (only one allowed)
+            FilenameNotExistsError: File has no filename
+            FileTooLargeError: File exceeds size limit
+            UnsupportedFileTypeError: File type not supported
+        """
         if "file" not in request.files:
             raise NoFileUploadedError()
 

api/controllers/web/forgot_password.py

@@ -16,7 +16,7 @@ from controllers.console.auth.error import (
 )
 from controllers.console.error import EmailSendIpLimitError
 from controllers.console.wraps import email_password_login_enabled, only_edition_enterprise, setup_required
-from controllers.web import api
+from controllers.web import web_ns
 from extensions.ext_database import db
 from libs.helper import email, extract_remote_ip
 from libs.password import hash_password, valid_password
@@ -24,10 +24,21 @@ from models.account import Account
 from services.account_service import AccountService
 
 
+@web_ns.route("/forgot-password")
 class ForgotPasswordSendEmailApi(Resource):
     @only_edition_enterprise
     @setup_required
     @email_password_login_enabled
+    @web_ns.doc("send_forgot_password_email")
+    @web_ns.doc(description="Send password reset email")
+    @web_ns.doc(
+        responses={
+            200: "Password reset email sent successfully",
+            400: "Bad request - invalid email format",
+            404: "Account not found",
+            429: "Too many requests - rate limit exceeded",
+        }
+    )
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=email, required=True, location="json")
@@ -54,10 +65,16 @@ class ForgotPasswordSendEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
+@web_ns.route("/forgot-password/validity")
 class ForgotPasswordCheckApi(Resource):
     @only_edition_enterprise
     @setup_required
     @email_password_login_enabled
+    @web_ns.doc("check_forgot_password_token")
+    @web_ns.doc(description="Verify password reset token validity")
+    @web_ns.doc(
+        responses={200: "Token is valid", 400: "Bad request - invalid token format", 401: "Invalid or expired token"}
+    )
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=str, required=True, location="json")
@@ -94,10 +111,21 @@ class ForgotPasswordCheckApi(Resource):
         return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
 
 
+@web_ns.route("/forgot-password/resets")
 class ForgotPasswordResetApi(Resource):
     @only_edition_enterprise
     @setup_required
     @email_password_login_enabled
+    @web_ns.doc("reset_password")
+    @web_ns.doc(description="Reset user password with verification token")
+    @web_ns.doc(
+        responses={
+            200: "Password reset successfully",
+            400: "Bad request - invalid parameters or password mismatch",
+            401: "Invalid or expired token",
+            404: "Account not found",
+        }
+    )
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("token", type=str, required=True, nullable=False, location="json")
@@ -141,8 +169,3 @@ class ForgotPasswordResetApi(Resource):
         account.password = base64.b64encode(password_hashed).decode()
         account.password_salt = base64.b64encode(salt).decode()
         session.commit()
-
-
-api.add_resource(ForgotPasswordSendEmailApi, "/forgot-password")
-api.add_resource(ForgotPasswordCheckApi, "/forgot-password/validity")
-api.add_resource(ForgotPasswordResetApi, "/forgot-password/resets")

api/controllers/web/login.py

@@ -1,5 +1,5 @@
 from flask_restx import Resource, reqparse
-from jwt import InvalidTokenError  # type: ignore
+from jwt import InvalidTokenError
 
 import services
 from controllers.console.auth.error import (
@@ -9,18 +9,30 @@ from controllers.console.auth.error import (
 )
 from controllers.console.error import AccountBannedError
 from controllers.console.wraps import only_edition_enterprise, setup_required
-from controllers.web import api
+from controllers.web import web_ns
 from libs.helper import email
 from libs.password import valid_password
 from services.account_service import AccountService
 from services.webapp_auth_service import WebAppAuthService
 
 
+@web_ns.route("/login")
 class LoginApi(Resource):
     """Resource for web app email/password login."""
 
     @setup_required
     @only_edition_enterprise
+    @web_ns.doc("web_app_login")
+    @web_ns.doc(description="Authenticate user for web application access")
+    @web_ns.doc(
+        responses={
+            200: "Authentication successful",
+            400: "Bad request - invalid email or password format",
+            401: "Authentication failed - email or password mismatch",
+            403: "Account banned or login disabled",
+            404: "Account not found",
+        }
+    )
     def post(self):
         """Authenticate user and login."""
         parser = reqparse.RequestParser()
@@ -51,9 +63,19 @@ class LoginApi(Resource):
 #         return {"result": "success"}
 
 
+@web_ns.route("/email-code-login")
 class EmailCodeLoginSendEmailApi(Resource):
     @setup_required
     @only_edition_enterprise
+    @web_ns.doc("send_email_code_login")
+    @web_ns.doc(description="Send email verification code for login")
+    @web_ns.doc(
+        responses={
+            200: "Email code sent successfully",
+            400: "Bad request - invalid email format",
+            404: "Account not found",
+        }
+    )
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=email, required=True, location="json")
@@ -74,9 +96,20 @@ class EmailCodeLoginSendEmailApi(Resource):
         return {"result": "success", "data": token}
 
 
+@web_ns.route("/email-code-login/validity")
 class EmailCodeLoginApi(Resource):
     @setup_required
     @only_edition_enterprise
+    @web_ns.doc("verify_email_code_login")
+    @web_ns.doc(description="Verify email code and complete login")
+    @web_ns.doc(
+        responses={
+            200: "Email code verified and login successful",
+            400: "Bad request - invalid code or token",
+            401: "Invalid token or expired code",
+            404: "Account not found",
+        }
+    )
     def post(self):
         parser = reqparse.RequestParser()
         parser.add_argument("email", type=str, required=True, location="json")
@@ -104,9 +137,3 @@ class EmailCodeLoginApi(Resource):
         token = WebAppAuthService.login(account=account)
         AccountService.reset_login_error_rate_limit(args["email"])
         return {"result": "success", "data": {"access_token": token}}
-
-
-api.add_resource(LoginApi, "/login")
-# api.add_resource(LogoutApi, "/logout")
-api.add_resource(EmailCodeLoginSendEmailApi, "/email-code-login")
-api.add_resource(EmailCodeLoginApi, "/email-code-login/validity")

api/controllers/web/message.py

@@ -85,6 +85,11 @@ class MessageListApi(WebApiResource):
 
 
 class MessageFeedbackApi(WebApiResource):
+    feedback_response_fields = {
+        "result": fields.String,
+    }
+
+    @marshal_with(feedback_response_fields)
     def post(self, app_model, end_user, message_id):
         message_id = str(message_id)
 
@@ -152,6 +157,11 @@ class MessageMoreLikeThisApi(WebApiResource):
 
 
 class MessageSuggestedQuestionApi(WebApiResource):
+    suggested_questions_response_fields = {
+        "data": fields.List(fields.String),
+    }
+
+    @marshal_with(suggested_questions_response_fields)
     def get(self, app_model, end_user, message_id):
         app_mode = AppMode.value_of(app_model.mode)
         if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:

api/controllers/web/passport.py

@@ -7,7 +7,7 @@ from sqlalchemy import func, select
 from werkzeug.exceptions import NotFound, Unauthorized
 
 from configs import dify_config
-from controllers.web import api
+from controllers.web import web_ns
 from controllers.web.error import WebAppAuthRequiredError
 from extensions.ext_database import db
 from libs.passport import PassportService
@@ -17,9 +17,19 @@ from services.feature_service import FeatureService
 from services.webapp_auth_service import WebAppAuthService, WebAppAuthType
 
 
+@web_ns.route("/passport")
 class PassportResource(Resource):
     """Base resource for passport."""
 
+    @web_ns.doc("get_passport")
+    @web_ns.doc(description="Get authentication passport for web application access")
+    @web_ns.doc(
+        responses={
+            200: "Passport retrieved successfully",
+            401: "Unauthorized - missing app code or invalid authentication",
+            404: "Application or user not found",
+        }
+    )
     def get(self):
         system_features = FeatureService.get_system_features()
         app_code = request.headers.get("X-App-Code")
@@ -94,9 +104,6 @@ class PassportResource(Resource):
         }
 
 
-api.add_resource(PassportResource, "/passport")
-
-
 def decode_enterprise_webapp_user_id(jwt_token: str | None):
     """
     Decode the enterprise user session from the Authorization header.

api/controllers/web/remote_files.py

@@ -10,16 +10,44 @@ from controllers.common.errors import (
     RemoteFileUploadError,
     UnsupportedFileTypeError,
 )
+from controllers.web import web_ns
 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 fields.file_fields import build_file_with_signed_url_model, build_remote_file_info_model
 from services.file_service import FileService
 
 
+@web_ns.route("/remote-files/<path:url>")
 class RemoteFileInfoApi(WebApiResource):
-    @marshal_with(remote_file_info_fields)
+    @web_ns.doc("get_remote_file_info")
+    @web_ns.doc(description="Get information about a remote file")
+    @web_ns.doc(
+        responses={
+            200: "Remote file information retrieved successfully",
+            400: "Bad request - invalid URL",
+            404: "Remote file not found",
+            500: "Failed to fetch remote file",
+        }
+    )
+    @marshal_with(build_remote_file_info_model(web_ns))
     def get(self, app_model, end_user, url):
+        """Get information about a remote file.
+
+        Retrieves basic information about a file located at a remote URL,
+        including content type and content length.
+
+        Args:
+            app_model: The associated application model
+            end_user: The end user making the request
+            url: URL-encoded path to the remote file
+
+        Returns:
+            dict: Remote file information including type and length
+
+        Raises:
+            HTTPException: If the remote file cannot be accessed
+        """
         decoded_url = urllib.parse.unquote(url)
         resp = ssrf_proxy.head(decoded_url)
         if resp.status_code != httpx.codes.OK:
@@ -32,9 +60,42 @@ class RemoteFileInfoApi(WebApiResource):
         }
 
 
+@web_ns.route("/remote-files/upload")
 class RemoteFileUploadApi(WebApiResource):
-    @marshal_with(file_fields_with_signed_url)
-    def post(self, app_model, end_user):  # Add app_model and end_user parameters
+    @web_ns.doc("upload_remote_file")
+    @web_ns.doc(description="Upload a file from a remote URL")
+    @web_ns.doc(
+        responses={
+            201: "Remote file uploaded successfully",
+            400: "Bad request - invalid URL or parameters",
+            413: "File too large",
+            415: "Unsupported file type",
+            500: "Failed to fetch remote file",
+        }
+    )
+    @marshal_with(build_file_with_signed_url_model(web_ns))
+    def post(self, app_model, end_user):
+        """Upload a file from a remote URL.
+
+        Downloads a file from the provided remote URL and uploads it
+        to the platform storage for use in web applications.
+
+        Args:
+            app_model: The associated application model
+            end_user: The end user making the request
+
+        JSON Parameters:
+            url: The remote URL to download the file from (required)
+
+        Returns:
+            dict: File information including ID, signed URL, and metadata
+            int: HTTP status code 201 for success
+
+        Raises:
+            RemoteFileUploadError: Failed to fetch file from remote URL
+            FileTooLargeError: File exceeds size limit
+            UnsupportedFileTypeError: File type not supported
+        """
         parser = reqparse.RequestParser()
         parser.add_argument("url", type=str, required=True, help="URL is required")
         args = parser.parse_args()

api/controllers/web/saved_message.py

@@ -30,6 +30,10 @@ class SavedMessageListApi(WebApiResource):
         "data": fields.List(fields.Nested(message_fields)),
     }
 
+    post_response_fields = {
+        "result": fields.String,
+    }
+
     @marshal_with(saved_message_infinite_scroll_pagination_fields)
     def get(self, app_model, end_user):
         if app_model.mode != "completion":
@@ -42,6 +46,7 @@ class SavedMessageListApi(WebApiResource):
 
         return SavedMessageService.pagination_by_last_id(app_model, end_user, args["last_id"], args["limit"])
 
+    @marshal_with(post_response_fields)
     def post(self, app_model, end_user):
         if app_model.mode != "completion":
             raise NotCompletionAppError()
@@ -59,6 +64,11 @@ class SavedMessageListApi(WebApiResource):
 
 
 class SavedMessageApi(WebApiResource):
+    delete_response_fields = {
+        "result": fields.String,
+    }
+
+    @marshal_with(delete_response_fields)
     def delete(self, app_model, end_user, message_id):
         message_id = str(message_id)
 

api/controllers/web/site.py

@@ -53,6 +53,18 @@ class AppSiteApi(WebApiResource):
         "custom_config": fields.Raw(attribute="custom_config"),
     }
 
+    @api.doc("Get App Site Info")
+    @api.doc(description="Retrieve app site information and configuration.")
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "App Not Found",
+            500: "Internal Server Error",
+        }
+    )
     @marshal_with(app_fields)
     def get(self, app_model, end_user):
         """Retrieve app site info."""

api/controllers/web/workflow.py

@@ -30,6 +30,24 @@ logger = logging.getLogger(__name__)
 
 
 class WorkflowRunApi(WebApiResource):
+    @api.doc("Run Workflow")
+    @api.doc(description="Execute a workflow with provided inputs and files.")
+    @api.doc(
+        params={
+            "inputs": {"description": "Input variables for the workflow", "type": "object", "required": True},
+            "files": {"description": "Files to be processed by the workflow", "type": "array", "required": False},
+        }
+    )
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "App Not Found",
+            500: "Internal Server Error",
+        }
+    )
     def post(self, app_model: App, end_user: EndUser):
         """
         Run workflow
@@ -67,6 +85,23 @@ class WorkflowRunApi(WebApiResource):
 
 
 class WorkflowTaskStopApi(WebApiResource):
+    @api.doc("Stop Workflow Task")
+    @api.doc(description="Stop a running workflow task.")
+    @api.doc(
+        params={
+            "task_id": {"description": "Task ID to stop", "type": "string", "required": True},
+        }
+    )
+    @api.doc(
+        responses={
+            200: "Success",
+            400: "Bad Request",
+            401: "Unauthorized",
+            403: "Forbidden",
+            404: "Task Not Found",
+            500: "Internal Server Error",
+        }
+    )
     def post(self, app_model: App, end_user: EndUser, task_id: str):
         """
         Stop workflow task

api/controllers/web/wraps.py

@@ -1,9 +1,11 @@
 from datetime import UTC, datetime
 from functools import wraps
+from typing import ParamSpec, TypeVar
 
 from flask import request
 from flask_restx import Resource
 from sqlalchemy import select
+from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
 
 from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
@@ -14,6 +16,9 @@ from services.enterprise.enterprise_service import EnterpriseService, WebAppSett
 from services.feature_service import FeatureService
 from services.webapp_auth_service import WebAppAuthService
 
+P = ParamSpec("P")
+R = TypeVar("R")
+
 
 def validate_jwt_token(view=None):
     def decorator(view):
@@ -49,18 +54,19 @@ def decode_jwt_token():
         decoded = PassportService().verify(tk)
         app_code = decoded.get("app_code")
         app_id = decoded.get("app_id")
-        app_model = db.session.scalar(select(App).where(App.id == app_id))
-        site = db.session.scalar(select(Site).where(Site.code == app_code))
-        if not app_model:
-            raise NotFound()
-        if not app_code or not site:
-            raise BadRequest("Site URL is no longer valid.")
-        if app_model.enable_site is False:
-            raise BadRequest("Site is disabled.")
-        end_user_id = decoded.get("end_user_id")
-        end_user = db.session.scalar(select(EndUser).where(EndUser.id == end_user_id))
-        if not end_user:
-            raise NotFound()
+        with Session(db.engine, expire_on_commit=False) as session:
+            app_model = session.scalar(select(App).where(App.id == app_id))
+            site = session.scalar(select(Site).where(Site.code == app_code))
+            if not app_model:
+                raise NotFound()
+            if not app_code or not site:
+                raise BadRequest("Site URL is no longer valid.")
+            if app_model.enable_site is False:
+                raise BadRequest("Site is disabled.")
+            end_user_id = decoded.get("end_user_id")
+            end_user = session.scalar(select(EndUser).where(EndUser.id == end_user_id))
+            if not end_user:
+                raise NotFound()
 
         # for enterprise webapp auth
         app_web_auth_enabled = False

api/core/agent/base_agent_runner.py

@@ -62,7 +62,7 @@ class BaseAgentRunner(AppRunner):
         model_instance: ModelInstance,
         memory: Optional[TokenBufferMemory] = None,
         prompt_messages: Optional[list[PromptMessage]] = None,
-    ) -> None:
+    ):
         self.tenant_id = tenant_id
         self.application_generate_entity = application_generate_entity
         self.conversation = conversation
@@ -334,7 +334,8 @@ class BaseAgentRunner(AppRunner):
         """
         Save agent thought
         """
-        agent_thought = db.session.query(MessageAgentThought).where(MessageAgentThought.id == agent_thought_id).first()
+        stmt = select(MessageAgentThought).where(MessageAgentThought.id == agent_thought_id)
+        agent_thought = db.session.scalar(stmt)
         if not agent_thought:
             raise ValueError("agent thought not found")
 
@@ -492,7 +493,8 @@ class BaseAgentRunner(AppRunner):
         return result
 
     def organize_agent_user_prompt(self, message: Message) -> UserPromptMessage:
-        files = db.session.query(MessageFile).where(MessageFile.message_id == message.id).all()
+        stmt = select(MessageFile).where(MessageFile.message_id == message.id)
+        files = db.session.scalars(stmt).all()
         if not files:
             return UserPromptMessage(content=message.query)
         if message.app_model_config:

api/core/agent/cot_agent_runner.py

@@ -338,7 +338,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
 
         return instruction
 
-    def _init_react_state(self, query) -> None:
+    def _init_react_state(self, query):
         """
         init agent scratchpad
         """

api/core/agent/entities.py

@@ -41,7 +41,7 @@ class AgentScratchpadUnit(BaseModel):
         action_name: str
         action_input: Union[dict, str]
 
-        def to_dict(self) -> dict:
+        def to_dict(self):
             """
             Convert to dictionary.
             """

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

@@ -158,7 +158,7 @@ class DatasetConfigManager:
         return config, ["agent_mode", "dataset_configs", "dataset_query_variable"]
 
     @classmethod
-    def extract_dataset_config_for_legacy_compatibility(cls, tenant_id: str, app_mode: AppMode, config: dict) -> dict:
+    def extract_dataset_config_for_legacy_compatibility(cls, tenant_id: str, app_mode: AppMode, config: dict):
         """
         Extract dataset config for legacy compatibility
 

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

@@ -105,7 +105,7 @@ class ModelConfigManager:
         return dict(config), ["model"]
 
     @classmethod
-    def validate_model_completion_params(cls, cp: dict) -> dict:
+    def validate_model_completion_params(cls, cp: dict):
         # model.completion_params
         if not isinstance(cp, dict):
             raise ValueError("model.completion_params must be of object type")

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

@@ -122,7 +122,7 @@ class PromptTemplateConfigManager:
         return config, ["prompt_type", "pre_prompt", "chat_prompt_config", "completion_prompt_config"]
 
     @classmethod
-    def validate_post_prompt_and_set_defaults(cls, config: dict) -> dict:
+    def validate_post_prompt_and_set_defaults(cls, config: dict):
         """
         Validate post_prompt and set defaults for prompt feature
 

api/core/app/app_config/features/more_like_this/manager.py

@@ -1,3 +1,16 @@
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+
+class MoreLikeThisConfig(BaseModel):
+    enabled: bool = False
+    model_config = ConfigDict(extra="allow")
+
+
+class AppConfigModel(BaseModel):
+    more_like_this: MoreLikeThisConfig = Field(default_factory=MoreLikeThisConfig)
+    model_config = ConfigDict(extra="allow")
+
+
 class MoreLikeThisConfigManager:
     @classmethod
     def convert(cls, config: dict) -> bool:
@@ -6,31 +19,14 @@ class MoreLikeThisConfigManager:
 
         :param config: model config args
         """
-        more_like_this = False
-        more_like_this_dict = config.get("more_like_this")
-        if more_like_this_dict:
-            if more_like_this_dict.get("enabled"):
-                more_like_this = True
-
-        return more_like_this
+        validated_config, _ = cls.validate_and_set_defaults(config)
+        return AppConfigModel.model_validate(validated_config).more_like_this.enabled
 
     @classmethod
     def validate_and_set_defaults(cls, config: dict) -> tuple[dict, list[str]]:
-        """
-        Validate and set defaults for more like this feature
-
-        :param config: app model config args
-        """
-        if not config.get("more_like_this"):
-            config["more_like_this"] = {"enabled": False}
-
-        if not isinstance(config["more_like_this"], dict):
-            raise ValueError("more_like_this must be of dict type")
-
-        if "enabled" not in config["more_like_this"] or not config["more_like_this"]["enabled"]:
-            config["more_like_this"]["enabled"] = False
-
-        if not isinstance(config["more_like_this"]["enabled"], bool):
-            raise ValueError("enabled in more_like_this must be of boolean type")
-
-        return config, ["more_like_this"]
+        try:
+            return AppConfigModel.model_validate(config).model_dump(), ["more_like_this"]
+        except ValidationError:
+            raise ValueError(
+                "more_like_this must be of dict type and enabled in more_like_this must be of boolean type"
+            )

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

@@ -41,7 +41,7 @@ class AdvancedChatAppConfigManager(BaseAppConfigManager):
         return app_config
 
     @classmethod
-    def config_validate(cls, tenant_id: str, config: dict, only_structure_validate: bool = False) -> dict:
+    def config_validate(cls, tenant_id: str, config: dict, only_structure_validate: bool = False):
         """
         Validate for advanced chat app model config
 

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

@@ -450,6 +450,12 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
 
         worker_thread.start()
 
+        # release database connection, because the following new thread operations may take a long time
+        db.session.refresh(workflow)
+        db.session.refresh(message)
+        # db.session.refresh(user)
+        db.session.close()
+
         # return response or stream generator
         response = self._handle_advanced_chat_response(
             application_generate_entity=application_generate_entity,
@@ -475,7 +481,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
         message_id: str,
         context: contextvars.Context,
         variable_loader: VariableLoader,
-    ) -> None:
+    ):
         """
         Generate worker in a new thread.
         :param flask_app: Flask app

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

@@ -54,7 +54,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         workflow: Workflow,
         system_user_id: str,
         app: App,
-    ) -> None:
+    ):
         super().__init__(
             queue_manager=queue_manager,
             variable_loader=variable_loader,
@@ -68,11 +68,13 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self.system_user_id = system_user_id
         self._app = app
 
-    def run(self) -> None:
+    def run(self):
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
-        app_record = db.session.query(App).where(App.id == app_config.app_id).first()
+        with Session(db.engine, expire_on_commit=False) as session:
+            app_record = session.scalar(select(App).where(App.id == app_config.app_id))
+
         if not app_record:
             raise ValueError("App not found")
 
@@ -140,7 +142,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 environment_variables=self._workflow.environment_variables,
                 # Based on the definition of `VariableUnion`,
                 # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
-                conversation_variables=cast(list[VariableUnion], conversation_variables),
+                conversation_variables=conversation_variables,
             )
 
             # init graph
@@ -219,7 +221,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
         return False
 
-    def _complete_with_stream_output(self, text: str, stopped_by: QueueStopEvent.StopBy) -> None:
+    def _complete_with_stream_output(self, text: str, stopped_by: QueueStopEvent.StopBy):
         """
         Direct output
         """

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

@@ -118,7 +118,7 @@ class AdvancedChatAppGenerateResponseConverter(AppGenerateResponseConverter):
                 data = cls._error_to_stream_response(sub_stream_response.err)
                 response_chunk.update(data)
             elif isinstance(sub_stream_response, NodeStartStreamResponse | NodeFinishStreamResponse):
-                response_chunk.update(sub_stream_response.to_ignore_detail_dict())
+                response_chunk.update(sub_stream_response.to_ignore_detail_dict())  # ty: ignore [unresolved-attribute]
             else:
                 response_chunk.update(sub_stream_response.to_dict())
 

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

@@ -1,4 +1,5 @@
 import logging
+import re
 import time
 from collections.abc import Callable, Generator, Mapping
 from contextlib import contextmanager
@@ -72,7 +73,6 @@ from core.workflow.repositories.workflow_execution_repository import WorkflowExe
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 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
@@ -101,7 +101,7 @@ class AdvancedChatAppGenerateTaskPipeline:
         workflow_execution_repository: WorkflowExecutionRepository,
         workflow_node_execution_repository: WorkflowNodeExecutionRepository,
         draft_var_saver_factory: DraftVariableSaverFactory,
-    ) -> None:
+    ):
         self._base_task_pipeline = BasedGenerateTaskPipeline(
             application_generate_entity=application_generate_entity,
             queue_manager=queue_manager,
@@ -143,6 +143,7 @@ class AdvancedChatAppGenerateTaskPipeline:
 
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
+            user=user,
         )
 
         self._task_state = WorkflowTaskState()
@@ -288,7 +289,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                 session.rollback()
                 raise
 
-    def _ensure_workflow_initialized(self) -> None:
+    def _ensure_workflow_initialized(self):
         """Fluent validation for workflow state."""
         if not self._workflow_run_id:
             raise ValueError("workflow run not initialized.")
@@ -309,13 +310,8 @@ class AdvancedChatAppGenerateTaskPipeline:
             err = self._base_task_pipeline._handle_error(event=event, session=session, message_id=self._message_id)
         yield self._base_task_pipeline._error_to_stream_response(err)
 
-    def _handle_workflow_started_event(
-        self, event: QueueWorkflowStartedEvent, *, graph_runtime_state: Optional[GraphRuntimeState] = None, **kwargs
-    ) -> Generator[StreamResponse, None, None]:
+    def _handle_workflow_started_event(self, *args, **kwargs) -> Generator[StreamResponse, None, None]:
         """Handle workflow started events."""
-        # Override graph runtime state - this is a side effect but necessary
-        graph_runtime_state = event.graph_runtime_state
-
         with self._database_session() as session:
             workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
             self._workflow_run_id = workflow_execution.id_
@@ -336,15 +332,14 @@ class AdvancedChatAppGenerateTaskPipeline:
         """Handle node retry events."""
         self._ensure_workflow_initialized()
 
-        with self._database_session() as session:
-            workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
-                workflow_execution_id=self._workflow_run_id, event=event
-            )
-            node_retry_resp = self._workflow_response_converter.workflow_node_retry_to_stream_response(
-                event=event,
-                task_id=self._application_generate_entity.task_id,
-                workflow_node_execution=workflow_node_execution,
-            )
+        workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
+            workflow_execution_id=self._workflow_run_id, event=event
+        )
+        node_retry_resp = self._workflow_response_converter.workflow_node_retry_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            workflow_node_execution=workflow_node_execution,
+        )
 
         if node_retry_resp:
             yield node_retry_resp
@@ -373,18 +368,17 @@ class AdvancedChatAppGenerateTaskPipeline:
     ) -> Generator[StreamResponse, None, None]:
         """Handle node succeeded events."""
         # Record files if it's an answer node or end node
-        if event.node_type in [NodeType.ANSWER, NodeType.END]:
+        if event.node_type in [NodeType.ANSWER, NodeType.END, NodeType.LLM]:
             self._recorded_files.extend(
                 self._workflow_response_converter.fetch_files_from_node_outputs(event.outputs or {})
             )
 
-        with self._database_session() as session:
-            workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_success(event=event)
-            node_finish_resp = self._workflow_response_converter.workflow_node_finish_to_stream_response(
-                event=event,
-                task_id=self._application_generate_entity.task_id,
-                workflow_node_execution=workflow_node_execution,
-            )
+        workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_success(event=event)
+        node_finish_resp = self._workflow_response_converter.workflow_node_finish_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            workflow_node_execution=workflow_node_execution,
+        )
 
         self._save_output_for_event(event, workflow_node_execution.id)
 
@@ -894,9 +888,16 @@ class AdvancedChatAppGenerateTaskPipeline:
         if self._conversation_name_generate_thread:
             self._conversation_name_generate_thread.join()
 
-    def _save_message(self, *, session: Session, graph_runtime_state: Optional[GraphRuntimeState] = None) -> None:
+    def _save_message(self, *, session: Session, graph_runtime_state: Optional[GraphRuntimeState] = None):
         message = self._get_message(session=session)
-        message.answer = self._task_state.answer
+
+        # If there are assistant files, remove markdown image links from answer
+        answer_text = self._task_state.answer
+        if self._recorded_files:
+            # Remove markdown image links since we're storing files separately
+            answer_text = re.sub(r"!\[.*?\]\(.*?\)", "", answer_text).strip()
+
+        message.answer = answer_text
         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()
@@ -930,10 +931,6 @@ class AdvancedChatAppGenerateTaskPipeline:
             self._task_state.metadata.usage = usage
         else:
             self._task_state.metadata.usage = LLMUsage.empty_usage()
-        message_was_created.send(
-            message,
-            application_generate_entity=self._application_generate_entity,
-        )
 
     def _message_end_to_stream_response(self) -> MessageEndStreamResponse:
         """

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

@@ -86,7 +86,7 @@ class AgentChatAppConfigManager(BaseAppConfigManager):
         return app_config
 
     @classmethod
-    def config_validate(cls, tenant_id: str, config: Mapping[str, Any]) -> dict:
+    def config_validate(cls, tenant_id: str, config: Mapping[str, Any]):
         """
         Validate for agent chat app model config
 

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

@@ -222,7 +222,7 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
         queue_manager: AppQueueManager,
         conversation_id: str,
         message_id: str,
-    ) -> None:
+    ):
         """
         Generate worker in a new thread.
         :param flask_app: Flask app

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

@@ -1,6 +1,8 @@
 import logging
 from typing import cast
 
+from sqlalchemy import select
+
 from core.agent.cot_chat_agent_runner import CotChatAgentRunner
 from core.agent.cot_completion_agent_runner import CotCompletionAgentRunner
 from core.agent.entities import AgentEntity
@@ -33,7 +35,7 @@ class AgentChatAppRunner(AppRunner):
         queue_manager: AppQueueManager,
         conversation: Conversation,
         message: Message,
-    ) -> None:
+    ):
         """
         Run assistant application
         :param application_generate_entity: application generate entity
@@ -44,8 +46,8 @@ class AgentChatAppRunner(AppRunner):
         """
         app_config = application_generate_entity.app_config
         app_config = cast(AgentChatAppConfig, app_config)
-
-        app_record = db.session.query(App).where(App.id == app_config.app_id).first()
+        app_stmt = select(App).where(App.id == app_config.app_id)
+        app_record = db.session.scalar(app_stmt)
         if not app_record:
             raise ValueError("App not found")
 
@@ -182,11 +184,12 @@ class AgentChatAppRunner(AppRunner):
 
         if {ModelFeature.MULTI_TOOL_CALL, ModelFeature.TOOL_CALL}.intersection(model_schema.features or []):
             agent_entity.strategy = AgentEntity.Strategy.FUNCTION_CALLING
-
-        conversation_result = db.session.query(Conversation).where(Conversation.id == conversation.id).first()
+        conversation_stmt = select(Conversation).where(Conversation.id == conversation.id)
+        conversation_result = db.session.scalar(conversation_stmt)
         if conversation_result is None:
             raise ValueError("Conversation not found")
-        message_result = db.session.query(Message).where(Message.id == message.id).first()
+        msg_stmt = select(Message).where(Message.id == message.id)
+        message_result = db.session.scalar(msg_stmt)
         if message_result is None:
             raise ValueError("Message not found")
         db.session.close()

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

@@ -16,7 +16,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
     _blocking_response_type = ChatbotAppBlockingResponse
 
     @classmethod
-    def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse) -> dict:  # type: ignore[override]
+    def convert_blocking_full_response(cls, blocking_response: ChatbotAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking full response.
         :param blocking_response: blocking response
@@ -37,7 +37,7 @@ class AgentChatAppGenerateResponseConverter(AppGenerateResponseConverter):
         return response
 
     @classmethod
-    def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse) -> dict:  # type: ignore[override]
+    def convert_blocking_simple_response(cls, blocking_response: ChatbotAppBlockingResponse):  # type: ignore[override]
         """
         Convert blocking simple response.
         :param blocking_response: blocking response

api/core/app/apps/base_app_generate_response_converter.py

@@ -94,7 +94,7 @@ class AppGenerateResponseConverter(ABC):
         return metadata
 
     @classmethod
-    def _error_to_stream_response(cls, e: Exception) -> dict:
+    def _error_to_stream_response(cls, e: Exception):
         """
         Error to stream response.
         :param e: exception

api/core/app/apps/base_app_generator.py

@@ -1,4 +1,3 @@
-import json
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Optional, Union, final
 
@@ -14,6 +13,7 @@ from core.workflow.repositories.draft_variable_repository import (
     NoopDraftVariableSaver,
 )
 from factories import file_factory
+from libs.orjson import orjson_dumps
 from services.workflow_draft_variable_service import DraftVariableSaver as DraftVariableSaverImpl
 
 if TYPE_CHECKING:
@@ -157,7 +157,7 @@ class BaseAppGenerator:
 
         return value
 
-    def _sanitize_value(self, value: Any) -> Any:
+    def _sanitize_value(self, value: Any):
         if isinstance(value, str):
             return value.replace("\x00", "")
         return value
@@ -174,7 +174,7 @@ class BaseAppGenerator:
             def gen():
                 for message in generator:
                     if isinstance(message, Mapping | dict):
-                        yield f"data: {json.dumps(message)}\n\n"
+                        yield f"data: {orjson_dumps(message)}\n\n"
                     else:
                         yield f"event: {message}\n\n"