chore(docker): bump version (#25092)

Signed-off-by: -LAN- <laipz8200@outlook.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
+  }

.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,14 @@ 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: 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 +93,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/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 }}

.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,7 +197,6 @@ sdks/python-client/dify_client.egg-info
 .vscode/*
 !.vscode/launch.json.template
 !.vscode/README.md
-pyrightconfig.json
 api/.vscode
 # vscode Code History Extension
 .history

.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/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/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/remote_settings_sources/nacos/__init__.py

@@ -29,7 +29,7 @@ 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
 

api/configs/remote_settings_sources/nacos/http_request.py

@@ -27,7 +27,7 @@ class NacosHttpClient:
             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"):
@@ -77,6 +77,6 @@ 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:
+        except Exception:
             logger.exception("[get-access-token] exception occur")
             raise

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

@@ -130,15 +130,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,7 +84,7 @@ 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)

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/message.py

@@ -95,18 +95,22 @@ class ChatMessageListApi(Resource):
                 .all()
             )
 
+        # Initialize has_more based on whether we have a full page
         if len(history_messages) == args["limit"]:
             current_page_first_message = history_messages[-1]
-
-        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,
+            # 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,
+                    )
                 )
             )
-        )
+        else:
+            # If we don't have a full page, there are no more messages
+            has_more = False
 
         history_messages = list(reversed(history_messages))
 
@@ -126,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/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,187 @@
+from functools import wraps
+from typing import cast
+
+import flask_login
+from flask import 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
+
+
+def oauth_server_client_id_required(view):
+    @wraps(view)
+    def decorated(*args, **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")
+
+        kwargs["oauth_provider_app"] = oauth_provider_app
+
+        return view(*args, **kwargs)
+
+    return decorated
+
+
+def oauth_server_access_token_required(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        oauth_provider_app = kwargs.get("oauth_provider_app")
+        if not oauth_provider_app or not isinstance(oauth_provider_app, OAuthProviderApp):
+            raise BadRequest("Invalid oauth_provider_app")
+
+        authorization_header = request.headers.get("Authorization")
+        if not authorization_header:
+            raise BadRequest("Authorization header is required")
+
+        parts = authorization_header.strip().split(" ")
+        if len(parts) != 2:
+            raise BadRequest("Invalid Authorization header format")
+
+        token_type = parts[0].strip()
+        if token_type.lower() != "bearer":
+            raise BadRequest("token_type is invalid")
+
+        access_token = parts[1].strip()
+        if not access_token:
+            raise BadRequest("access_token is required")
+
+        account = OAuthServerService.validate_oauth_access_token(oauth_provider_app.client_id, access_token)
+        if not account:
+            raise BadRequest("access_token or client_id is invalid")
+
+        kwargs["account"] = account
+
+        return view(*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/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()
@@ -488,13 +486,13 @@ 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":
                 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"],
@@ -506,7 +504,7 @@ class DocumentBatchIndexingEstimateApi(DocumentResource):
                 extract_settings.append(extract_setting)
             elif document.data_source_type == "website_crawl":
                 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/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/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) -> None:
+        """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/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 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
@@ -23,14 +23,14 @@ from models.model import ApiToken, App, EndUser
 from services.feature_service import FeatureService
 
 
-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):
@@ -291,27 +291,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/conversation.py

@@ -73,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
 
 

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 (

api/controllers/web/wraps.py

@@ -4,6 +4,7 @@ from functools import wraps
 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
@@ -49,18 +50,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

@@ -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/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_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,

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

@@ -72,7 +72,9 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         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

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
@@ -143,6 +143,7 @@ class AdvancedChatAppGenerateTaskPipeline:
 
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
+            user=user,
         )
 
         self._task_state = WorkflowTaskState()
@@ -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)
 
@@ -896,7 +890,14 @@ class AdvancedChatAppGenerateTaskPipeline:
 
     def _save_message(self, *, session: Session, graph_runtime_state: Optional[GraphRuntimeState] = None) -> None:
         message = self._get_message(session=session)
-        message.answer = self._task_state.answer
+
+        # 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_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
@@ -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/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:
@@ -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"
 

api/core/app/apps/base_app_queue_manager.py

@@ -1,7 +1,7 @@
 import queue
 import time
 from abc import abstractmethod
-from enum import Enum
+from enum import IntEnum, auto
 from typing import Any, Optional
 
 from sqlalchemy.orm import DeclarativeMeta
@@ -19,9 +19,9 @@ from core.app.entities.queue_entities import (
 from extensions.ext_redis import redis_client
 
 
-class PublishFrom(Enum):
-    APPLICATION_MANAGER = 1
-    TASK_PIPELINE = 2
+class PublishFrom(IntEnum):
+    APPLICATION_MANAGER = auto()
+    TASK_PIPELINE = auto()
 
 
 class AppQueueManager:
@@ -159,7 +159,7 @@ class AppQueueManager:
     def _check_for_sqlalchemy_models(self, data: Any):
         # from entity to dict or list
         if isinstance(data, dict):
-            for key, value in data.items():
+            for value in data.values():
                 self._check_for_sqlalchemy_models(value)
         elif isinstance(data, list):
             for item in data:

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

@@ -1,6 +1,8 @@
 import logging
 from typing import cast
 
+from sqlalchemy import select
+
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.base_app_runner import AppRunner
 from core.app.apps.chat.app_config_manager import ChatAppConfig
@@ -42,8 +44,8 @@ class ChatAppRunner(AppRunner):
         """
         app_config = application_generate_entity.app_config
         app_config = cast(ChatAppConfig, app_config)
-
-        app_record = db.session.query(App).where(App.id == app_config.app_id).first()
+        stmt = select(App).where(App.id == app_config.app_id)
+        app_record = db.session.scalar(stmt)
         if not app_record:
             raise ValueError("App not found")
 

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

@@ -3,7 +3,6 @@ from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime
 from typing import Any, Optional, Union, cast
 
-from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
@@ -53,9 +52,7 @@ from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from libs.datetime_utils import naive_utc_now
 from models import (
     Account,
-    CreatorUserRole,
     EndUser,
-    WorkflowRun,
 )
 
 
@@ -64,8 +61,10 @@ class WorkflowResponseConverter:
         self,
         *,
         application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
+        user: Union[Account, EndUser],
     ) -> None:
         self._application_generate_entity = application_generate_entity
+        self._user = user
 
     def workflow_start_to_stream_response(
         self,
@@ -92,27 +91,21 @@ class WorkflowResponseConverter:
         workflow_execution: WorkflowExecution,
     ) -> WorkflowFinishStreamResponse:
         created_by = None
-        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id_))
-        assert workflow_run is not None
-        if workflow_run.created_by_role == CreatorUserRole.ACCOUNT:
-            stmt = select(Account).where(Account.id == workflow_run.created_by)
-            account = session.scalar(stmt)
-            if account:
-                created_by = {
-                    "id": account.id,
-                    "name": account.name,
-                    "email": account.email,
-                }
-        elif workflow_run.created_by_role == CreatorUserRole.END_USER:
-            stmt = select(EndUser).where(EndUser.id == workflow_run.created_by)
-            end_user = session.scalar(stmt)
-            if end_user:
-                created_by = {
-                    "id": end_user.id,
-                    "user": end_user.session_id,
-                }
+
+        user = self._user
+        if isinstance(user, Account):
+            created_by = {
+                "id": user.id,
+                "name": user.name,
+                "email": user.email,
+            }
+        elif isinstance(user, EndUser):
+            created_by = {
+                "id": user.id,
+                "user": user.session_id,
+            }
         else:
-            raise NotImplementedError(f"unknown created_by_role: {workflow_run.created_by_role}")
+            raise NotImplementedError(f"User type not supported: {type(user)}")
 
         # Handle the case where finished_at is None by using current time as default
         finished_at_timestamp = (

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

@@ -6,6 +6,7 @@ from typing import Any, Literal, Union, overload
 
 from flask import Flask, copy_current_request_context, current_app
 from pydantic import ValidationError
+from sqlalchemy import select
 
 from configs import dify_config
 from core.app.app_config.easy_ui_based_app.model_config.converter import ModelConfigConverter
@@ -248,17 +249,14 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
         :param invoke_from: invoke from source
         :param stream: is stream
         """
-        message = (
-            db.session.query(Message)
-            .where(
-                Message.id == message_id,
-                Message.app_id == app_model.id,
-                Message.from_source == ("api" if isinstance(user, EndUser) else "console"),
-                Message.from_end_user_id == (user.id if isinstance(user, EndUser) else None),
-                Message.from_account_id == (user.id if isinstance(user, Account) else None),
-            )
-            .first()
+        stmt = select(Message).where(
+            Message.id == message_id,
+            Message.app_id == app_model.id,
+            Message.from_source == ("api" if isinstance(user, EndUser) else "console"),
+            Message.from_end_user_id == (user.id if isinstance(user, EndUser) else None),
+            Message.from_account_id == (user.id if isinstance(user, Account) else None),
         )
+        message = db.session.scalar(stmt)
 
         if not message:
             raise MessageNotExistsError()

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

@@ -1,6 +1,8 @@
 import logging
 from typing import cast
 
+from sqlalchemy import select
+
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.apps.base_app_runner import AppRunner
 from core.app.apps.completion.app_config_manager import CompletionAppConfig
@@ -35,8 +37,8 @@ class CompletionAppRunner(AppRunner):
         """
         app_config = application_generate_entity.app_config
         app_config = cast(CompletionAppConfig, app_config)
-
-        app_record = db.session.query(App).where(App.id == app_config.app_id).first()
+        stmt = select(App).where(App.id == app_config.app_id)
+        app_record = db.session.scalar(stmt)
         if not app_record:
             raise ValueError("App not found")
 

api/core/app/apps/message_based_app_generator.py

@@ -3,6 +3,9 @@ import logging
 from collections.abc import Generator
 from typing import Optional, Union, cast
 
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
 from core.app.app_config.entities import EasyUIBasedAppConfig, EasyUIBasedAppModelConfigFrom
 from core.app.apps.base_app_generator import BaseAppGenerator
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -83,11 +86,10 @@ class MessageBasedAppGenerator(BaseAppGenerator):
 
     def _get_app_model_config(self, app_model: App, conversation: Optional[Conversation] = None) -> AppModelConfig:
         if conversation:
-            app_model_config = (
-                db.session.query(AppModelConfig)
-                .where(AppModelConfig.id == conversation.app_model_config_id, AppModelConfig.app_id == app_model.id)
-                .first()
+            stmt = select(AppModelConfig).where(
+                AppModelConfig.id == conversation.app_model_config_id, AppModelConfig.app_id == app_model.id
             )
+            app_model_config = db.session.scalar(stmt)
 
             if not app_model_config:
                 raise AppModelConfigBrokenError()
@@ -253,7 +255,8 @@ class MessageBasedAppGenerator(BaseAppGenerator):
         :param conversation_id: conversation id
         :return: conversation
         """
-        conversation = db.session.query(Conversation).where(Conversation.id == conversation_id).first()
+        with Session(db.engine, expire_on_commit=False) as session:
+            conversation = session.scalar(select(Conversation).where(Conversation.id == conversation_id))
 
         if not conversation:
             raise ConversationNotExistsError("Conversation not exists")
@@ -266,7 +269,8 @@ class MessageBasedAppGenerator(BaseAppGenerator):
         :param message_id: message id
         :return: message
         """
-        message = db.session.query(Message).where(Message.id == message_id).first()
+        with Session(db.engine, expire_on_commit=False) as session:
+            message = session.scalar(select(Message).where(Message.id == message_id))
 
         if message is None:
             raise MessageNotExistsError("Message not exists")

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

@@ -89,7 +89,7 @@ class WorkflowAppGenerateResponseConverter(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())
             yield response_chunk

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

@@ -131,6 +131,7 @@ class WorkflowAppGenerateTaskPipeline:
 
         self._workflow_response_converter = WorkflowResponseConverter(
             application_generate_entity=application_generate_entity,
+            user=user,
         )
 
         self._application_generate_entity = application_generate_entity
@@ -299,16 +300,15 @@ class WorkflowAppGenerateTaskPipeline:
         """Handle node retry events."""
         self._ensure_workflow_initialized()
 
-        with self._database_session() as session:
-            workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
-                workflow_execution_id=self._workflow_run_id,
-                event=event,
-            )
-            response = self._workflow_response_converter.workflow_node_retry_to_stream_response(
-                event=event,
-                task_id=self._application_generate_entity.task_id,
-                workflow_node_execution=workflow_node_execution,
-            )
+        workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
+            workflow_execution_id=self._workflow_run_id,
+            event=event,
+        )
+        response = self._workflow_response_converter.workflow_node_retry_to_stream_response(
+            event=event,
+            task_id=self._application_generate_entity.task_id,
+            workflow_node_execution=workflow_node_execution,
+        )
 
         if response:
             yield response

api/core/app/entities/app_invoke_entities.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping, Sequence
-from enum import Enum
+from enum import StrEnum
 from typing import Any, Optional
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
@@ -11,7 +11,7 @@ from core.file import File, FileUploadConfig
 from core.model_runtime.entities.model_entities import AIModelEntity
 
 
-class InvokeFrom(Enum):
+class InvokeFrom(StrEnum):
     """
     Invoke From.
     """

api/core/app/entities/queue_entities.py

@@ -118,7 +118,7 @@ class QueueIterationNextEvent(AppQueueEvent):
     parent_parallel_start_node_id: Optional[str] = None
     """parent parallel start node id if node is in parallel"""
     parallel_mode_run_id: Optional[str] = None
-    """iteratoin run in parallel mode run id"""
+    """iteration run in parallel mode run id"""
     node_run_index: int
     output: Optional[Any] = None  # output for the current iteration
     duration: Optional[float] = None
@@ -201,7 +201,7 @@ class QueueLoopNextEvent(AppQueueEvent):
     parent_parallel_start_node_id: Optional[str] = None
     """parent parallel start node id if node is in parallel"""
     parallel_mode_run_id: Optional[str] = None
-    """iteratoin run in parallel mode run id"""
+    """iteration run in parallel mode run id"""
     node_run_index: int
     output: Optional[Any] = None  # output for the current loop
     duration: Optional[float] = None
@@ -382,7 +382,7 @@ class QueueNodeStartedEvent(AppQueueEvent):
     """loop id if node is in loop"""
     start_at: datetime
     parallel_mode_run_id: Optional[str] = None
-    """iteratoin run in parallel mode run id"""
+    """iteration run in parallel mode run id"""
     agent_strategy: Optional[AgentNodeStrategyInit] = None
 
 

api/core/app/features/annotation_reply/annotation_reply.py

@@ -1,6 +1,8 @@
 import logging
 from typing import Optional
 
+from sqlalchemy import select
+
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.rag.datasource.vdb.vector_factory import Vector
 from extensions.ext_database import db
@@ -25,9 +27,8 @@ class AnnotationReplyFeature:
         :param invoke_from: invoke from
         :return:
         """
-        annotation_setting = (
-            db.session.query(AppAnnotationSetting).where(AppAnnotationSetting.app_id == app_record.id).first()
-        )
+        stmt = select(AppAnnotationSetting).where(AppAnnotationSetting.app_id == app_record.id)
+        annotation_setting = db.session.scalar(stmt)
 
         if not annotation_setting:
             return None

api/core/app/features/rate_limiting/rate_limit.py

@@ -96,7 +96,11 @@ class RateLimit:
         if isinstance(generator, Mapping):
             return generator
         else:
-            return RateLimitGenerator(rate_limit=self, generator=generator, request_id=request_id)
+            return RateLimitGenerator(
+                rate_limit=self,
+                generator=generator,  # ty: ignore [invalid-argument-type]
+                request_id=request_id,
+            )
 
 
 class RateLimitGenerator:

api/core/app/task_pipeline/based_generate_task_pipeline.py

@@ -50,7 +50,7 @@ class BasedGenerateTaskPipeline:
         if isinstance(e, InvokeAuthorizationError):
             err = InvokeAuthorizationError("Incorrect API key provided")
         elif isinstance(e, InvokeError | ValueError):
-            err = e
+            err = e  # ty: ignore [invalid-assignment]
         else:
             description = getattr(e, "description", None)
             err = Exception(description if description is not None else str(e))

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -472,9 +472,10 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
         :param event: agent thought event
         :return:
         """
-        agent_thought: Optional[MessageAgentThought] = (
-            db.session.query(MessageAgentThought).where(MessageAgentThought.id == event.agent_thought_id).first()
-        )
+        with Session(db.engine, expire_on_commit=False) as session:
+            agent_thought: Optional[MessageAgentThought] = (
+                session.query(MessageAgentThought).where(MessageAgentThought.id == event.agent_thought_id).first()
+            )
 
         if agent_thought:
             return AgentThoughtStreamResponse(

api/core/app/task_pipeline/exc.py

@@ -1,8 +1,8 @@
-class TaskPipilineError(ValueError):
+class TaskPipelineError(ValueError):
     pass
 
 
-class RecordNotFoundError(TaskPipilineError):
+class RecordNotFoundError(TaskPipelineError):
     def __init__(self, record_name: str, record_id: str):
         super().__init__(f"{record_name} with id {record_id} not found")
 

api/core/app/task_pipeline/message_cycle_manager.py

@@ -3,6 +3,8 @@ from threading import Thread
 from typing import Optional, Union
 
 from flask import Flask, current_app
+from sqlalchemy import select
+from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import (
@@ -84,7 +86,8 @@ class MessageCycleManager:
     def _generate_conversation_name_worker(self, flask_app: Flask, conversation_id: str, query: str):
         with flask_app.app_context():
             # get conversation and message
-            conversation = db.session.query(Conversation).where(Conversation.id == conversation_id).first()
+            stmt = select(Conversation).where(Conversation.id == conversation_id)
+            conversation = db.session.scalar(stmt)
 
             if not conversation:
                 return
@@ -98,7 +101,7 @@ class MessageCycleManager:
                 try:
                     name = LLMGenerator.generate_conversation_name(app_model.tenant_id, query)
                     conversation.name = name
-                except Exception as e:
+                except Exception:
                     if dify_config.DEBUG:
                         logger.exception("generate conversation name failed, conversation_id: %s", conversation_id)
                     pass
@@ -143,7 +146,8 @@ class MessageCycleManager:
         :param event: event
         :return:
         """
-        message_file = db.session.query(MessageFile).where(MessageFile.id == event.message_file_id).first()
+        with Session(db.engine, expire_on_commit=False) as session:
+            message_file = session.scalar(select(MessageFile).where(MessageFile.id == event.message_file_id))
 
         if message_file and message_file.url is not None:
             # get tool file id
@@ -183,7 +187,8 @@ class MessageCycleManager:
         :param message_id: message id
         :return:
         """
-        message_file = db.session.query(MessageFile).where(MessageFile.id == message_id).first()
+        with Session(db.engine, expire_on_commit=False) as session:
+            message_file = session.scalar(select(MessageFile).where(MessageFile.id == message_id))
         event_type = StreamEvent.MESSAGE_FILE if message_file else StreamEvent.MESSAGE
 
         return MessageStreamResponse(

api/core/callback_handler/index_tool_callback_handler.py

@@ -1,6 +1,8 @@
 import logging
 from collections.abc import Sequence
 
+from sqlalchemy import select
+
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import QueueRetrieverResourcesEvent
@@ -49,7 +51,8 @@ class DatasetIndexToolCallbackHandler:
         for document in documents:
             if document.metadata is not None:
                 document_id = document.metadata["document_id"]
-                dataset_document = db.session.query(DatasetDocument).where(DatasetDocument.id == document_id).first()
+                dataset_document_stmt = select(DatasetDocument).where(DatasetDocument.id == document_id)
+                dataset_document = db.session.scalar(dataset_document_stmt)
                 if not dataset_document:
                     _logger.warning(
                         "Expected DatasetDocument record to exist, but none was found, document_id=%s",
@@ -57,17 +60,14 @@ class DatasetIndexToolCallbackHandler:
                     )
                     continue
                 if dataset_document.doc_form == IndexType.PARENT_CHILD_INDEX:
-                    child_chunk = (
-                        db.session.query(ChildChunk)
-                        .where(
-                            ChildChunk.index_node_id == document.metadata["doc_id"],
-                            ChildChunk.dataset_id == dataset_document.dataset_id,
-                            ChildChunk.document_id == dataset_document.id,
-                        )
-                        .first()
+                    child_chunk_stmt = select(ChildChunk).where(
+                        ChildChunk.index_node_id == document.metadata["doc_id"],
+                        ChildChunk.dataset_id == dataset_document.dataset_id,
+                        ChildChunk.document_id == dataset_document.id,
                     )
+                    child_chunk = db.session.scalar(child_chunk_stmt)
                     if child_chunk:
-                        segment = (
+                        _ = (
                             db.session.query(DocumentSegment)
                             .where(DocumentSegment.id == child_chunk.segment_id)
                             .update(

api/core/entities/provider_configuration.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import re
 from collections import defaultdict
 from collections.abc import Iterator, Sequence
 from json import JSONDecodeError
@@ -343,7 +344,65 @@ class ProviderConfiguration(BaseModel):
             with Session(db.engine) as new_session:
                 return _validate(new_session)
 
-    def create_provider_credential(self, credentials: dict, credential_name: str) -> None:
+    def _generate_provider_credential_name(self, session) -> str:
+        """
+        Generate a unique credential name for provider.
+        :return: credential name
+        """
+        return self._generate_next_api_key_name(
+            session=session,
+            query_factory=lambda: select(ProviderCredential).where(
+                ProviderCredential.tenant_id == self.tenant_id,
+                ProviderCredential.provider_name == self.provider.provider,
+            ),
+        )
+
+    def _generate_custom_model_credential_name(self, model: str, model_type: ModelType, session) -> str:
+        """
+        Generate a unique credential name for custom model.
+        :return: credential name
+        """
+        return self._generate_next_api_key_name(
+            session=session,
+            query_factory=lambda: select(ProviderModelCredential).where(
+                ProviderModelCredential.tenant_id == self.tenant_id,
+                ProviderModelCredential.provider_name == self.provider.provider,
+                ProviderModelCredential.model_name == model,
+                ProviderModelCredential.model_type == model_type.to_origin_model_type(),
+            ),
+        )
+
+    def _generate_next_api_key_name(self, session, query_factory) -> str:
+        """
+        Generate next available API KEY name by finding the highest numbered suffix.
+        :param session: database session
+        :param query_factory: function that returns the SQLAlchemy query
+        :return: next available API KEY name
+        """
+        try:
+            stmt = query_factory()
+            credential_records = session.execute(stmt).scalars().all()
+
+            if not credential_records:
+                return "API KEY 1"
+
+            # Extract numbers from API KEY pattern using list comprehension
+            pattern = re.compile(r"^API KEY\s+(\d+)$")
+            numbers = [
+                int(match.group(1))
+                for cr in credential_records
+                if cr.credential_name and (match := pattern.match(cr.credential_name.strip()))
+            ]
+
+            # Return next sequential number
+            next_number = max(numbers, default=0) + 1
+            return f"API KEY {next_number}"
+
+        except Exception as e:
+            logger.warning("Error generating next credential name: %s", str(e))
+            return "API KEY 1"
+
+    def create_provider_credential(self, credentials: dict, credential_name: str | None) -> None:
         """
         Add custom provider credentials.
         :param credentials: provider credentials
@@ -351,8 +410,11 @@ class ProviderConfiguration(BaseModel):
         :return:
         """
         with Session(db.engine) as session:
-            if self._check_provider_credential_name_exists(credential_name=credential_name, session=session):
-                raise ValueError(f"Credential with name '{credential_name}' already exists.")
+            if credential_name:
+                if self._check_provider_credential_name_exists(credential_name=credential_name, session=session):
+                    raise ValueError(f"Credential with name '{credential_name}' already exists.")
+            else:
+                credential_name = self._generate_provider_credential_name(session)
 
             credentials = self.validate_provider_credentials(credentials=credentials, session=session)
             provider_record = self._get_provider_record(session)
@@ -395,7 +457,7 @@ class ProviderConfiguration(BaseModel):
         self,
         credentials: dict,
         credential_id: str,
-        credential_name: str,
+        credential_name: str | None,
     ) -> None:
         """
         update a saved provider credential (by credential_id).
@@ -406,7 +468,7 @@ class ProviderConfiguration(BaseModel):
         :return:
         """
         with Session(db.engine) as session:
-            if self._check_provider_credential_name_exists(
+            if credential_name and self._check_provider_credential_name_exists(
                 credential_name=credential_name, session=session, exclude_id=credential_id
             ):
                 raise ValueError(f"Credential with name '{credential_name}' already exists.")
@@ -428,9 +490,9 @@ class ProviderConfiguration(BaseModel):
             try:
                 # Update credential
                 credential_record.encrypted_config = json.dumps(credentials)
-                credential_record.credential_name = credential_name
                 credential_record.updated_at = naive_utc_now()
-
+                if credential_name:
+                    credential_record.credential_name = credential_name
                 session.commit()
 
                 if provider_record and provider_record.credential_id == credential_id:
@@ -532,13 +594,7 @@ class ProviderConfiguration(BaseModel):
                         cache_type=ProviderCredentialsCacheType.LOAD_BALANCING_MODEL,
                     )
                     lb_credentials_cache.delete()
-
-                    lb_config.credential_id = None
-                    lb_config.encrypted_config = None
-                    lb_config.enabled = False
-                    lb_config.name = "__delete__"
-                    lb_config.updated_at = naive_utc_now()
-                    session.add(lb_config)
+                    session.delete(lb_config)
 
                 # Check if this is the currently active credential
                 provider_record = self._get_provider_record(session)
@@ -822,7 +878,7 @@ class ProviderConfiguration(BaseModel):
                 return _validate(new_session)
 
     def create_custom_model_credential(
-        self, model_type: ModelType, model: str, credentials: dict, credential_name: str
+        self, model_type: ModelType, model: str, credentials: dict, credential_name: str | None
     ) -> None:
         """
         Create a custom model credential.
@@ -833,10 +889,15 @@ class ProviderConfiguration(BaseModel):
         :return:
         """
         with Session(db.engine) as session:
-            if self._check_custom_model_credential_name_exists(
-                model=model, model_type=model_type, credential_name=credential_name, session=session
-            ):
-                raise ValueError(f"Model credential with name '{credential_name}' already exists for {model}.")
+            if credential_name:
+                if self._check_custom_model_credential_name_exists(
+                    model=model, model_type=model_type, credential_name=credential_name, session=session
+                ):
+                    raise ValueError(f"Model credential with name '{credential_name}' already exists for {model}.")
+            else:
+                credential_name = self._generate_custom_model_credential_name(
+                    model=model, model_type=model_type, session=session
+                )
             # validate custom model config
             credentials = self.validate_custom_model_credentials(
                 model_type=model_type, model=model, credentials=credentials, session=session
@@ -880,7 +941,7 @@ class ProviderConfiguration(BaseModel):
                 raise
 
     def update_custom_model_credential(
-        self, model_type: ModelType, model: str, credentials: dict, credential_name: str, credential_id: str
+        self, model_type: ModelType, model: str, credentials: dict, credential_name: str | None, credential_id: str
     ) -> None:
         """
         Update a custom model credential.
@@ -893,7 +954,7 @@ class ProviderConfiguration(BaseModel):
         :return:
         """
         with Session(db.engine) as session:
-            if self._check_custom_model_credential_name_exists(
+            if credential_name and self._check_custom_model_credential_name_exists(
                 model=model,
                 model_type=model_type,
                 credential_name=credential_name,
@@ -925,8 +986,9 @@ class ProviderConfiguration(BaseModel):
             try:
                 # Update credential
                 credential_record.encrypted_config = json.dumps(credentials)
-                credential_record.credential_name = credential_name
                 credential_record.updated_at = naive_utc_now()
+                if credential_name:
+                    credential_record.credential_name = credential_name
                 session.commit()
 
                 if provider_model_record and provider_model_record.credential_id == credential_id:
@@ -982,12 +1044,7 @@ class ProviderConfiguration(BaseModel):
                         cache_type=ProviderCredentialsCacheType.LOAD_BALANCING_MODEL,
                     )
                     lb_credentials_cache.delete()
-                    lb_config.credential_id = None
-                    lb_config.encrypted_config = None
-                    lb_config.enabled = False
-                    lb_config.name = "__delete__"
-                    lb_config.updated_at = naive_utc_now()
-                    session.add(lb_config)
+                    session.delete(lb_config)
 
                 # Check if this is the currently active credential
                 provider_model_record = self._get_custom_model_record(model_type, model, session=session)
@@ -1054,6 +1111,7 @@ class ProviderConfiguration(BaseModel):
                     provider_name=self.provider.provider,
                     model_name=model,
                     model_type=model_type.to_origin_model_type(),
+                    is_valid=True,
                     credential_id=credential_id,
                 )
             else:
@@ -1605,11 +1663,9 @@ class ProviderConfiguration(BaseModel):
                         if config.credential_source_type != "custom_model"
                     ]
 
-                    if len(provider_model_lb_configs) > 1:
-                        load_balancing_enabled = True
-
-                    if any(config.name == "__delete__" for config in provider_model_lb_configs):
-                        has_invalid_load_balancing_configs = True
+                    load_balancing_enabled = model_setting.load_balancing_enabled
+                    # when the user enable load_balancing but available configs are less than 2 display warning
+                    has_invalid_load_balancing_configs = load_balancing_enabled and len(provider_model_lb_configs) < 2
 
                 provider_models.append(
                     ModelWithProviderEntity(
@@ -1631,6 +1687,8 @@ class ProviderConfiguration(BaseModel):
         for model_configuration in self.custom_configuration.models:
             if model_configuration.model_type not in model_types:
                 continue
+            if model_configuration.unadded_to_model_list:
+                continue
             if model and model != model_configuration.model:
                 continue
             try:
@@ -1663,11 +1721,9 @@ class ProviderConfiguration(BaseModel):
                     if config.credential_source_type != "provider"
                 ]
 
-                if len(custom_model_lb_configs) > 1:
-                    load_balancing_enabled = True
-
-                if any(config.name == "__delete__" for config in custom_model_lb_configs):
-                    has_invalid_load_balancing_configs = True
+                load_balancing_enabled = model_setting.load_balancing_enabled
+                # when the user enable load_balancing but available configs are less than 2 display warning
+                has_invalid_load_balancing_configs = load_balancing_enabled and len(custom_model_lb_configs) < 2
 
             if len(model_configuration.available_model_credentials) > 0 and not model_configuration.credentials:
                 status = ModelStatus.CREDENTIAL_REMOVED

api/core/entities/provider_entities.py

@@ -111,11 +111,21 @@ class CustomModelConfiguration(BaseModel):
     current_credential_id: Optional[str] = None
     current_credential_name: Optional[str] = None
     available_model_credentials: list[CredentialConfiguration] = []
+    unadded_to_model_list: Optional[bool] = False
 
     # pydantic configs
     model_config = ConfigDict(protected_namespaces=())
 
 
+class UnaddedModelConfiguration(BaseModel):
+    """
+    Model class for provider unadded model configuration.
+    """
+
+    model: str
+    model_type: ModelType
+
+
 class CustomConfiguration(BaseModel):
     """
     Model class for provider custom configuration.
@@ -123,6 +133,7 @@ class CustomConfiguration(BaseModel):
 
     provider: Optional[CustomProviderConfiguration] = None
     models: list[CustomModelConfiguration] = []
+    can_added_models: list[UnaddedModelConfiguration] = []
 
 
 class ModelLoadBalancingConfiguration(BaseModel):
@@ -144,6 +155,7 @@ class ModelSettings(BaseModel):
     model: str
     model_type: ModelType
     enabled: bool = True
+    load_balancing_enabled: bool = False
     load_balancing_configs: list[ModelLoadBalancingConfiguration] = []
 
     # pydantic configs

api/core/extension/api_based_extension_requestor.py

@@ -43,9 +43,9 @@ class APIBasedExtensionRequestor:
                 timeout=self.timeout,
                 proxies=proxies,
             )
-        except requests.exceptions.Timeout:
+        except requests.Timeout:
             raise ValueError("request timeout")
-        except requests.exceptions.ConnectionError:
+        except requests.ConnectionError:
             raise ValueError("request connection error")
 
         if response.status_code != 200:

api/core/extension/extensible.py

@@ -91,7 +91,7 @@ class Extensible:
 
                 # Find extension class
                 extension_class = None
-                for name, obj in vars(mod).items():
+                for obj in vars(mod).values():
                     if isinstance(obj, type) and issubclass(obj, cls) and obj != cls:
                         extension_class = obj
                         break
@@ -123,7 +123,7 @@ class Extensible:
                     )
                 )
 
-        except Exception as e:
+        except Exception:
             logger.exception("Error scanning extensions")
             raise
 

api/core/extension/extension.py

@@ -41,9 +41,3 @@ class Extension:
         assert module_extension.extension_class is not None
         t: type = module_extension.extension_class
         return t
-
-    def validate_form_schema(self, module: ExtensionModule, extension_name: str, config: dict) -> None:
-        module_extension = self.module_extension(module, extension_name)
-        form_schema = module_extension.form_schema
-
-        # TODO validate form_schema

api/core/external_data_tool/api/api.py

@@ -1,5 +1,7 @@
 from typing import Optional
 
+from sqlalchemy import select
+
 from core.extension.api_based_extension_requestor import APIBasedExtensionRequestor
 from core.external_data_tool.base import ExternalDataTool
 from core.helper import encrypter
@@ -28,13 +30,11 @@ class ApiExternalDataTool(ExternalDataTool):
         api_based_extension_id = config.get("api_based_extension_id")
         if not api_based_extension_id:
             raise ValueError("api_based_extension_id is required")
-
         # get api_based_extension
-        api_based_extension = (
-            db.session.query(APIBasedExtension)
-            .where(APIBasedExtension.tenant_id == tenant_id, APIBasedExtension.id == api_based_extension_id)
-            .first()
+        stmt = select(APIBasedExtension).where(
+            APIBasedExtension.tenant_id == tenant_id, APIBasedExtension.id == api_based_extension_id
         )
+        api_based_extension = db.session.scalar(stmt)
 
         if not api_based_extension:
             raise ValueError("api_based_extension_id is invalid")
@@ -52,13 +52,11 @@ class ApiExternalDataTool(ExternalDataTool):
             raise ValueError(f"config is required, config: {self.config}")
         api_based_extension_id = self.config.get("api_based_extension_id")
         assert api_based_extension_id is not None, "api_based_extension_id is required"
-
         # get api_based_extension
-        api_based_extension = (
-            db.session.query(APIBasedExtension)
-            .where(APIBasedExtension.tenant_id == self.tenant_id, APIBasedExtension.id == api_based_extension_id)
-            .first()
+        stmt = select(APIBasedExtension).where(
+            APIBasedExtension.tenant_id == self.tenant_id, APIBasedExtension.id == api_based_extension_id
         )
+        api_based_extension = db.session.scalar(stmt)
 
         if not api_based_extension:
             raise ValueError(

api/core/external_data_tool/factory.py

@@ -22,7 +22,6 @@ class ExternalDataToolFactory:
         :param config: the form config data
         :return:
         """
-        code_based_extension.validate_form_schema(ExtensionModule.EXTERNAL_DATA_TOOL, name, config)
         extension_class = code_based_extension.extension_class(ExtensionModule.EXTERNAL_DATA_TOOL, name)
         # FIXME mypy issue here, figure out how to fix it
         extension_class.validate_config(tenant_id, config)  # type: ignore

api/core/file/file_manager.py

@@ -88,6 +88,7 @@ def to_prompt_message_content(
         "url": _to_url(f) if dify_config.MULTIMODAL_SEND_FORMAT == "url" else "",
         "format": f.extension.removeprefix("."),
         "mime_type": f.mime_type,
+        "filename": f.filename or "",
     }
     if f.type == FileType.IMAGE:
         params["detail"] = image_detail_config or ImagePromptMessageContent.DETAIL.LOW

api/core/helper/encrypter.py

@@ -3,7 +3,7 @@ import base64
 from libs import rsa
 
 
-def obfuscated_token(token: str):
+def obfuscated_token(token: str) -> str:
     if not token:
         return token
     if len(token) <= 8:
@@ -11,6 +11,10 @@ def obfuscated_token(token: str):
     return token[:6] + "*" * 12 + token[-2:]
 
 
+def full_mask_token(token_length=20):
+    return "*" * token_length
+
+
 def encrypt_token(tenant_id: str, token: str):
     from models.account import Tenant
     from models.engine import db

api/core/helper/marketplace.py

@@ -1,6 +1,6 @@
 from collections.abc import Sequence
 
-import requests
+import httpx
 from yarl import URL
 
 from configs import dify_config
@@ -23,7 +23,7 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
         return []
 
     url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = requests.post(url, json={"plugin_ids": plugin_ids})
+    response = httpx.post(url, json={"plugin_ids": plugin_ids})
     response.raise_for_status()
 
     return [MarketplacePluginDeclaration(**plugin) for plugin in response.json()["data"]["plugins"]]
@@ -36,13 +36,13 @@ def batch_fetch_plugin_manifests_ignore_deserialization_error(
         return []
 
     url = str(marketplace_api_url / "api/v1/plugins/batch")
-    response = requests.post(url, json={"plugin_ids": plugin_ids})
+    response = httpx.post(url, json={"plugin_ids": plugin_ids})
     response.raise_for_status()
     result: list[MarketplacePluginDeclaration] = []
     for plugin in response.json()["data"]["plugins"]:
         try:
             result.append(MarketplacePluginDeclaration(**plugin))
-        except Exception as e:
+        except Exception:
             pass
 
     return result
@@ -50,5 +50,5 @@ def batch_fetch_plugin_manifests_ignore_deserialization_error(
 
 def record_install_plugin_event(plugin_unique_identifier: str):
     url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
-    response = requests.post(url, json={"unique_identifier": plugin_unique_identifier})
+    response = httpx.post(url, json={"unique_identifier": plugin_unique_identifier})
     response.raise_for_status()

api/core/helper/module_import_helper.py

@@ -47,7 +47,7 @@ def get_subclasses_from_module(mod: ModuleType, parent_type: type) -> list[type]
 
 
 def load_single_subclass_from_source(
-    *, module_name: str, script_path: AnyStr, parent_type: type, use_lazy_loader: bool = False
+    *, module_name: str, script_path: str, parent_type: type, use_lazy_loader: bool = False
 ) -> type:
     """
     Load a single subclass from the source

api/core/helper/position_helper.py

@@ -1,7 +1,7 @@
 import os
 from collections import OrderedDict
 from collections.abc import Callable
-from typing import Any
+from typing import TypeVar
 
 from configs import dify_config
 from core.tools.utils.yaml_utils import load_yaml_file
@@ -72,11 +72,14 @@ def pin_position_map(original_position_map: dict[str, int], pin_list: list[str])
     return position_map
 
 
+T = TypeVar("T")
+
+
 def is_filtered(
     include_set: set[str],
     exclude_set: set[str],
-    data: Any,
-    name_func: Callable[[Any], str],
+    data: T,
+    name_func: Callable[[T], str],
 ) -> bool:
     """
     Check if the object should be filtered out.
@@ -103,9 +106,9 @@ def is_filtered(
 
 def sort_by_position_map(
     position_map: dict[str, int],
-    data: list[Any],
-    name_func: Callable[[Any], str],
-) -> list[Any]:
+    data: list[T],
+    name_func: Callable[[T], str],
+):
     """
     Sort the objects by the position map.
     If the name of the object is not in the position map, it will be put at the end.
@@ -122,9 +125,9 @@ def sort_by_position_map(
 
 def sort_to_dict_by_position_map(
     position_map: dict[str, int],
-    data: list[Any],
-    name_func: Callable[[Any], str],
-) -> OrderedDict[str, Any]:
+    data: list[T],
+    name_func: Callable[[T], str],
+):
     """
     Sort the objects into a ordered dict by the position map.
     If the name of the object is not in the position map, it will be put at the end.
@@ -134,4 +137,4 @@ def sort_to_dict_by_position_map(
     :return: an OrderedDict with the sorted pairs of name and object
     """
     sorted_items = sort_by_position_map(position_map, data, name_func)
-    return OrderedDict([(name_func(item), item) for item in sorted_items])
+    return OrderedDict((name_func(item), item) for item in sorted_items)

api/core/indexing_runner.py

@@ -5,9 +5,10 @@ import re
 import threading
 import time
 import uuid
-from typing import Any, Optional, cast
+from typing import Any, Optional
 
 from flask import current_app
+from sqlalchemy import select
 from sqlalchemy.orm.exc import ObjectDeletedError
 
 from configs import dify_config
@@ -18,6 +19,7 @@ from core.model_runtime.entities.model_entities import ModelType
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.docstore.dataset_docstore import DatasetDocumentStore
+from core.rag.extractor.entity.datasource_type import DatasourceType
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.index_processor.constant.index_type import IndexType
 from core.rag.index_processor.index_processor_base import BaseIndexProcessor
@@ -56,13 +58,11 @@ class IndexingRunner:
 
                 if not dataset:
                     raise ValueError("no dataset found")
-
                 # get the process rule
-                processing_rule = (
-                    db.session.query(DatasetProcessRule)
-                    .where(DatasetProcessRule.id == dataset_document.dataset_process_rule_id)
-                    .first()
+                stmt = select(DatasetProcessRule).where(
+                    DatasetProcessRule.id == dataset_document.dataset_process_rule_id
                 )
+                processing_rule = db.session.scalar(stmt)
                 if not processing_rule:
                     raise ValueError("no process rule found")
                 index_type = dataset_document.doc_form
@@ -123,11 +123,8 @@ class IndexingRunner:
                     db.session.query(ChildChunk).where(ChildChunk.segment_id == document_segment.id).delete()
             db.session.commit()
             # get the process rule
-            processing_rule = (
-                db.session.query(DatasetProcessRule)
-                .where(DatasetProcessRule.id == dataset_document.dataset_process_rule_id)
-                .first()
-            )
+            stmt = select(DatasetProcessRule).where(DatasetProcessRule.id == dataset_document.dataset_process_rule_id)
+            processing_rule = db.session.scalar(stmt)
             if not processing_rule:
                 raise ValueError("no process rule found")
 
@@ -208,7 +205,6 @@ class IndexingRunner:
                                     child_documents.append(child_document)
                                 document.children = child_documents
                         documents.append(document)
-
             # build index
             index_type = dataset_document.doc_form
             index_processor = IndexProcessorFactory(index_type).init_index_processor()
@@ -310,7 +306,8 @@ class IndexingRunner:
                 # delete image files and related db records
                 image_upload_file_ids = get_image_upload_file_ids(document.page_content)
                 for upload_file_id in image_upload_file_ids:
-                    image_file = db.session.query(UploadFile).where(UploadFile.id == upload_file_id).first()
+                    stmt = select(UploadFile).where(UploadFile.id == upload_file_id)
+                    image_file = db.session.scalar(stmt)
                     if image_file is None:
                         continue
                     try:
@@ -339,14 +336,14 @@ class IndexingRunner:
         if dataset_document.data_source_type == "upload_file":
             if not data_source_info or "upload_file_id" not in data_source_info:
                 raise ValueError("no upload file found")
-
-            file_detail = (
-                db.session.query(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"]).one_or_none()
-            )
+            stmt = select(UploadFile).where(UploadFile.id == data_source_info["upload_file_id"])
+            file_detail = db.session.scalars(stmt).one_or_none()
 
             if file_detail:
                 extract_setting = ExtractSetting(
-                    datasource_type="upload_file", upload_file=file_detail, document_model=dataset_document.doc_form
+                    datasource_type=DatasourceType.FILE.value,
+                    upload_file=file_detail,
+                    document_model=dataset_document.doc_form,
                 )
                 text_docs = index_processor.extract(extract_setting, process_rule_mode=process_rule["mode"])
         elif dataset_document.data_source_type == "notion_import":
@@ -357,7 +354,7 @@ class IndexingRunner:
             ):
                 raise ValueError("no notion import info found")
             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"],
@@ -377,7 +374,7 @@ class IndexingRunner:
             ):
                 raise ValueError("no website import info found")
             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"],
@@ -400,7 +397,6 @@ class IndexingRunner:
         )
 
         # replace doc id to document model id
-        text_docs = cast(list[Document], text_docs)
         for text_doc in text_docs:
             if text_doc.metadata is not None:
                 text_doc.metadata["document_id"] = dataset_document.id

api/core/llm_generator/llm_generator.py

@@ -56,11 +56,8 @@ class LLMGenerator:
         prompts = [UserPromptMessage(content=prompt)]
 
         with measure_time() as timer:
-            response = cast(
-                LLMResult,
-                model_instance.invoke_llm(
-                    prompt_messages=list(prompts), model_parameters={"max_tokens": 500, "temperature": 1}, stream=False
-                ),
+            response: LLMResult = model_instance.invoke_llm(
+                prompt_messages=list(prompts), model_parameters={"max_tokens": 500, "temperature": 1}, stream=False
             )
         answer = cast(str, response.message.content)
         cleaned_answer = re.sub(r"^.*(\{.*\}).*$", r"\1", answer, flags=re.DOTALL)
@@ -69,7 +66,7 @@ class LLMGenerator:
         try:
             result_dict = json.loads(cleaned_answer)
             answer = result_dict["Your Output"]
-        except json.JSONDecodeError as e:
+        except json.JSONDecodeError:
             logger.exception("Failed to generate name after answer, use query instead")
             answer = query
         name = answer.strip()
@@ -113,13 +110,10 @@ class LLMGenerator:
         prompt_messages = [UserPromptMessage(content=prompt)]
 
         try:
-            response = cast(
-                LLMResult,
-                model_instance.invoke_llm(
-                    prompt_messages=list(prompt_messages),
-                    model_parameters={"max_tokens": 256, "temperature": 0},
-                    stream=False,
-                ),
+            response: LLMResult = model_instance.invoke_llm(
+                prompt_messages=list(prompt_messages),
+                model_parameters={"max_tokens": 256, "temperature": 0},
+                stream=False,
             )
 
             text_content = response.message.get_text_content()
@@ -162,11 +156,8 @@ class LLMGenerator:
             )
 
             try:
-                response = cast(
-                    LLMResult,
-                    model_instance.invoke_llm(
-                        prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-                    ),
+                response: LLMResult = model_instance.invoke_llm(
+                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
                 )
 
                 rule_config["prompt"] = cast(str, response.message.content)
@@ -212,11 +203,8 @@ class LLMGenerator:
         try:
             try:
                 # the first step to generate the task prompt
-                prompt_content = cast(
-                    LLMResult,
-                    model_instance.invoke_llm(
-                        prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-                    ),
+                prompt_content: LLMResult = model_instance.invoke_llm(
+                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
                 )
             except InvokeError as e:
                 error = str(e)
@@ -248,11 +236,8 @@ class LLMGenerator:
             statement_messages = [UserPromptMessage(content=statement_generate_prompt)]
 
             try:
-                parameter_content = cast(
-                    LLMResult,
-                    model_instance.invoke_llm(
-                        prompt_messages=list(parameter_messages), model_parameters=model_parameters, stream=False
-                    ),
+                parameter_content: LLMResult = model_instance.invoke_llm(
+                    prompt_messages=list(parameter_messages), model_parameters=model_parameters, stream=False
                 )
                 rule_config["variables"] = re.findall(r'"\s*([^"]+)\s*"', cast(str, parameter_content.message.content))
             except InvokeError as e:
@@ -260,11 +245,8 @@ class LLMGenerator:
                 error_step = "generate variables"
 
             try:
-                statement_content = cast(
-                    LLMResult,
-                    model_instance.invoke_llm(
-                        prompt_messages=list(statement_messages), model_parameters=model_parameters, stream=False
-                    ),
+                statement_content: LLMResult = model_instance.invoke_llm(
+                    prompt_messages=list(statement_messages), model_parameters=model_parameters, stream=False
                 )
                 rule_config["opening_statement"] = cast(str, statement_content.message.content)
             except InvokeError as e:
@@ -307,11 +289,8 @@ class LLMGenerator:
         prompt_messages = [UserPromptMessage(content=prompt)]
         model_parameters = model_config.get("completion_params", {})
         try:
-            response = cast(
-                LLMResult,
-                model_instance.invoke_llm(
-                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-                ),
+            response: LLMResult = model_instance.invoke_llm(
+                prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
             )
 
             generated_code = cast(str, response.message.content)
@@ -338,13 +317,10 @@ class LLMGenerator:
 
         prompt_messages = [SystemPromptMessage(content=prompt), UserPromptMessage(content=query)]
 
-        response = cast(
-            LLMResult,
-            model_instance.invoke_llm(
-                prompt_messages=prompt_messages,
-                model_parameters={"temperature": 0.01, "max_tokens": 2000},
-                stream=False,
-            ),
+        response: LLMResult = model_instance.invoke_llm(
+            prompt_messages=prompt_messages,
+            model_parameters={"temperature": 0.01, "max_tokens": 2000},
+            stream=False,
         )
 
         answer = cast(str, response.message.content)
@@ -367,11 +343,8 @@ class LLMGenerator:
         model_parameters = model_config.get("model_parameters", {})
 
         try:
-            response = cast(
-                LLMResult,
-                model_instance.invoke_llm(
-                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-                ),
+            response: LLMResult = model_instance.invoke_llm(
+                prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
             )
 
             raw_content = response.message.content
@@ -555,11 +528,8 @@ class LLMGenerator:
         model_parameters = {"temperature": 0.4}
 
         try:
-            response = cast(
-                LLMResult,
-                model_instance.invoke_llm(
-                    prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
-                ),
+            response: LLMResult = model_instance.invoke_llm(
+                prompt_messages=list(prompt_messages), model_parameters=model_parameters, stream=False
             )
 
             generated_raw = cast(str, response.message.content)

api/core/mcp/auth/auth_flow.py

@@ -101,7 +101,7 @@ def handle_callback(state_key: str, authorization_code: str) -> OAuthCallbackSta
 
 def check_support_resource_discovery(server_url: str) -> tuple[bool, str]:
     """Check if the server supports OAuth 2.0 Resource Discovery."""
-    b_scheme, b_netloc, b_path, b_params, b_query, b_fragment = urlparse(server_url, "", True)
+    b_scheme, b_netloc, b_path, _, b_query, b_fragment = urlparse(server_url, "", True)
     url_for_resource_discovery = f"{b_scheme}://{b_netloc}/.well-known/oauth-protected-resource{b_path}"
     if b_query:
         url_for_resource_discovery += f"?{b_query}"
@@ -117,7 +117,7 @@ def check_support_resource_discovery(server_url: str) -> tuple[bool, str]:
             else:
                 return False, ""
         return False, ""
-    except httpx.RequestError as e:
+    except httpx.RequestError:
         # Not support resource discovery, fall back to well-known OAuth metadata
         return False, ""
 

api/core/mcp/client/streamable_client.py

@@ -246,6 +246,10 @@ class StreamableHTTPTransport:
                 logger.debug("Received 202 Accepted")
                 return
 
+            if response.status_code == 204:
+                logger.debug("Received 204 No Content")
+                return
+
             if response.status_code == 404:
                 if isinstance(message.root, JSONRPCRequest):
                     self._send_session_terminated_error(

api/core/mcp/mcp_client.py

@@ -2,7 +2,7 @@ import logging
 from collections.abc import Callable
 from contextlib import AbstractContextManager, ExitStack
 from types import TracebackType
-from typing import Any, Optional, cast
+from typing import Any, Optional
 from urllib.parse import urlparse
 
 from core.mcp.client.sse_client import sse_client
@@ -116,8 +116,7 @@ class MCPClient:
 
             self._session_context = ClientSession(*streams)
             self._session = self._exit_stack.enter_context(self._session_context)
-            session = cast(ClientSession, self._session)
-            session.initialize()
+            self._session.initialize()
             return
 
         except MCPAuthError:

api/core/mcp/server/streamable_http.py

@@ -4,224 +4,259 @@ from collections.abc import Mapping
 from typing import Any, cast
 
 from configs import dify_config
-from controllers.web.passport import generate_session_id
 from core.app.app_config.entities import VariableEntity, VariableEntityType
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
-from core.mcp import types
-from core.mcp.types import INTERNAL_ERROR, INVALID_PARAMS, METHOD_NOT_FOUND
-from core.mcp.utils import create_mcp_error_response
-from core.model_runtime.utils.encoders import jsonable_encoder
-from extensions.ext_database import db
+from core.mcp import types as mcp_types
 from models.model import App, AppMCPServer, AppMode, EndUser
 from services.app_generate_service import AppGenerateService
 
 logger = logging.getLogger(__name__)
 
 
-class MCPServerStreamableHTTPRequestHandler:
+def handle_mcp_request(
+    app: App,
+    request: mcp_types.ClientRequest,
+    user_input_form: list[VariableEntity],
+    mcp_server: AppMCPServer,
+    end_user: EndUser | None = None,
+    request_id: int | str = 1,
+) -> mcp_types.JSONRPCResponse | mcp_types.JSONRPCError:
     """
-    Apply to MCP HTTP streamable server with stateless http
+    Handle MCP request and return JSON-RPC response
+
+    Args:
+        app: The Dify app instance
+        request: The JSON-RPC request message
+        user_input_form: List of variable entities for the app
+        mcp_server: The MCP server configuration
+        end_user: Optional end user
+        request_id: The request ID
+
+    Returns:
+        JSON-RPC response or error
     """
 
-    def __init__(
-        self, app: App, request: types.ClientRequest | types.ClientNotification, user_input_form: list[VariableEntity]
-    ):
-        self.app = app
-        self.request = request
-        mcp_server = db.session.query(AppMCPServer).where(AppMCPServer.app_id == self.app.id).first()
-        if not mcp_server:
-            raise ValueError("MCP server not found")
-        self.mcp_server: AppMCPServer = mcp_server
-        self.end_user = self.retrieve_end_user()
-        self.user_input_form = user_input_form
+    request_type = type(request.root)
 
-    @property
-    def request_type(self):
-        return type(self.request.root)
+    def create_success_response(result_data: mcp_types.Result) -> mcp_types.JSONRPCResponse:
+        """Create success response with business result data"""
+        return mcp_types.JSONRPCResponse(
+            jsonrpc="2.0",
+            id=request_id,
+            result=result_data.model_dump(by_alias=True, mode="json", exclude_none=True),
+        )
 
-    @property
-    def parameter_schema(self):
-        parameters, required = self._convert_input_form_to_parameters(self.user_input_form)
-        if self.app.mode in {AppMode.COMPLETION.value, AppMode.WORKFLOW.value}:
-            return {
-                "type": "object",
-                "properties": parameters,
-                "required": required,
-            }
+    def create_error_response(code: int, message: str) -> mcp_types.JSONRPCError:
+        """Create error response with error code and message"""
+        from core.mcp.types import ErrorData
+
+        error_data = ErrorData(code=code, message=message)
+        return mcp_types.JSONRPCError(
+            jsonrpc="2.0",
+            id=request_id,
+            error=error_data,
+        )
+
+    # Request handler mapping using functional approach
+    request_handlers = {
+        mcp_types.InitializeRequest: lambda: handle_initialize(mcp_server.description),
+        mcp_types.ListToolsRequest: lambda: handle_list_tools(
+            app.name, app.mode, user_input_form, mcp_server.description, mcp_server.parameters_dict
+        ),
+        mcp_types.CallToolRequest: lambda: handle_call_tool(app, request, user_input_form, end_user),
+        mcp_types.PingRequest: lambda: handle_ping(),
+    }
+
+    try:
+        # Dispatch request to appropriate handler
+        handler = request_handlers.get(request_type)
+        if handler:
+            return create_success_response(handler())
+        else:
+            return create_error_response(mcp_types.METHOD_NOT_FOUND, f"Method not found: {request_type.__name__}")
+
+    except ValueError as e:
+        logger.exception("Invalid params")
+        return create_error_response(mcp_types.INVALID_PARAMS, str(e))
+    except Exception as e:
+        logger.exception("Internal server error")
+        return create_error_response(mcp_types.INTERNAL_ERROR, "Internal server error: " + str(e))
+
+
+def handle_ping() -> mcp_types.EmptyResult:
+    """Handle ping request"""
+    return mcp_types.EmptyResult()
+
+
+def handle_initialize(description: str) -> mcp_types.InitializeResult:
+    """Handle initialize request"""
+    capabilities = mcp_types.ServerCapabilities(
+        tools=mcp_types.ToolsCapability(listChanged=False),
+    )
+
+    return mcp_types.InitializeResult(
+        protocolVersion=mcp_types.SERVER_LATEST_PROTOCOL_VERSION,
+        capabilities=capabilities,
+        serverInfo=mcp_types.Implementation(name="Dify", version=dify_config.project.version),
+        instructions=description,
+    )
+
+
+def handle_list_tools(
+    app_name: str,
+    app_mode: str,
+    user_input_form: list[VariableEntity],
+    description: str,
+    parameters_dict: dict[str, str],
+) -> mcp_types.ListToolsResult:
+    """Handle list tools request"""
+    parameter_schema = build_parameter_schema(app_mode, user_input_form, parameters_dict)
+
+    return mcp_types.ListToolsResult(
+        tools=[
+            mcp_types.Tool(
+                name=app_name,
+                description=description,
+                inputSchema=parameter_schema,
+            )
+        ],
+    )
+
+
+def handle_call_tool(
+    app: App,
+    request: mcp_types.ClientRequest,
+    user_input_form: list[VariableEntity],
+    end_user: EndUser | None,
+) -> mcp_types.CallToolResult:
+    """Handle call tool request"""
+    request_obj = cast(mcp_types.CallToolRequest, request.root)
+    args = prepare_tool_arguments(app, request_obj.params.arguments or {})
+
+    if not end_user:
+        raise ValueError("End user not found")
+
+    response = AppGenerateService.generate(
+        app,
+        end_user,
+        args,
+        InvokeFrom.SERVICE_API,
+        streaming=app.mode == AppMode.AGENT_CHAT.value,
+    )
+
+    answer = extract_answer_from_response(app, response)
+    return mcp_types.CallToolResult(content=[mcp_types.TextContent(text=answer, type="text")])
+
+
+def build_parameter_schema(
+    app_mode: str,
+    user_input_form: list[VariableEntity],
+    parameters_dict: dict[str, str],
+) -> dict[str, Any]:
+    """Build parameter schema for the tool"""
+    parameters, required = convert_input_form_to_parameters(user_input_form, parameters_dict)
+
+    if app_mode in {AppMode.COMPLETION.value, AppMode.WORKFLOW.value}:
         return {
             "type": "object",
-            "properties": {
-                "query": {"type": "string", "description": "User Input/Question content"},
-                **parameters,
-            },
-            "required": ["query", *required],
+            "properties": parameters,
+            "required": required,
         }
+    return {
+        "type": "object",
+        "properties": {
+            "query": {"type": "string", "description": "User Input/Question content"},
+            **parameters,
+        },
+        "required": ["query", *required],
+    }
 
-    @property
-    def capabilities(self):
-        return types.ServerCapabilities(
-            tools=types.ToolsCapability(listChanged=False),
-        )
 
-    def response(self, response: types.Result | str):
-        if isinstance(response, str):
-            sse_content = f"event: ping\ndata: {response}\n\n".encode()
-            yield sse_content
-            return
-        json_response = types.JSONRPCResponse(
-            jsonrpc="2.0",
-            id=(self.request.root.model_extra or {}).get("id", 1),
-            result=response.model_dump(by_alias=True, mode="json", exclude_none=True),
-        )
-        json_data = json.dumps(jsonable_encoder(json_response))
+def prepare_tool_arguments(app: App, arguments: dict[str, Any]) -> dict[str, Any]:
+    """Prepare arguments based on app mode"""
+    if app.mode == AppMode.WORKFLOW.value:
+        return {"inputs": arguments}
+    elif app.mode == AppMode.COMPLETION.value:
+        return {"query": "", "inputs": arguments}
+    else:
+        # Chat modes - create a copy to avoid modifying original dict
+        args_copy = arguments.copy()
+        query = args_copy.pop("query", "")
+        return {"query": query, "inputs": args_copy}
 
-        sse_content = f"event: message\ndata: {json_data}\n\n".encode()
 
-        yield sse_content
+def extract_answer_from_response(app: App, response: Any) -> str:
+    """Extract answer from app generate response"""
+    answer = ""
 
-    def error_response(self, code: int, message: str, data=None):
-        request_id = (self.request.root.model_extra or {}).get("id", 1) or 1
-        return create_mcp_error_response(request_id, code, message, data)
+    if isinstance(response, RateLimitGenerator):
+        answer = process_streaming_response(response)
+    elif isinstance(response, Mapping):
+        answer = process_mapping_response(app, response)
+    else:
+        logger.warning("Unexpected response type: %s", type(response))
 
-    def handle(self):
-        handle_map = {
-            types.InitializeRequest: self.initialize,
-            types.ListToolsRequest: self.list_tools,
-            types.CallToolRequest: self.invoke_tool,
-            types.InitializedNotification: self.handle_notification,
-            types.PingRequest: self.handle_ping,
-        }
-        try:
-            if self.request_type in handle_map:
-                return self.response(handle_map[self.request_type]())
-            else:
-                return self.error_response(METHOD_NOT_FOUND, f"Method not found: {self.request_type}")
-        except ValueError as e:
-            logger.exception("Invalid params")
-            return self.error_response(INVALID_PARAMS, str(e))
-        except Exception as e:
-            logger.exception("Internal server error")
-            return self.error_response(INTERNAL_ERROR, f"Internal server error: {str(e)}")
+    return answer
 
-    def handle_notification(self):
-        return "ping"
 
-    def handle_ping(self):
-        return types.EmptyResult()
-
-    def initialize(self):
-        request = cast(types.InitializeRequest, self.request.root)
-        client_info = request.params.clientInfo
-        client_name = f"{client_info.name}@{client_info.version}"
-        if not self.end_user:
-            end_user = EndUser(
-                tenant_id=self.app.tenant_id,
-                app_id=self.app.id,
-                type="mcp",
-                name=client_name,
-                session_id=generate_session_id(),
-                external_user_id=self.mcp_server.id,
-            )
-            db.session.add(end_user)
-            db.session.commit()
-        return types.InitializeResult(
-            protocolVersion=types.SERVER_LATEST_PROTOCOL_VERSION,
-            capabilities=self.capabilities,
-            serverInfo=types.Implementation(name="Dify", version=dify_config.project.version),
-            instructions=self.mcp_server.description,
-        )
-
-    def list_tools(self):
-        if not self.end_user:
-            raise ValueError("User not found")
-        return types.ListToolsResult(
-            tools=[
-                types.Tool(
-                    name=self.app.name,
-                    description=self.mcp_server.description,
-                    inputSchema=self.parameter_schema,
-                )
-            ],
-        )
-
-    def invoke_tool(self):
-        if not self.end_user:
-            raise ValueError("User not found")
-        request = cast(types.CallToolRequest, self.request.root)
-        args = request.params.arguments or {}
-        if self.app.mode in {AppMode.WORKFLOW.value}:
-            args = {"inputs": args}
-        elif self.app.mode in {AppMode.COMPLETION.value}:
-            args = {"query": "", "inputs": args}
-        else:
-            args = {"query": args["query"], "inputs": {k: v for k, v in args.items() if k != "query"}}
-        response = AppGenerateService.generate(
-            self.app,
-            self.end_user,
-            args,
-            InvokeFrom.SERVICE_API,
-            streaming=self.app.mode == AppMode.AGENT_CHAT.value,
-        )
-        answer = ""
-        if isinstance(response, RateLimitGenerator):
-            for item in response.generator:
-                data = item
-                if isinstance(data, str) and data.startswith("data: "):
-                    try:
-                        json_str = data[6:].strip()
-                        parsed_data = json.loads(json_str)
-                        if parsed_data.get("event") == "agent_thought":
-                            answer += parsed_data.get("thought", "")
-                    except json.JSONDecodeError:
-                        continue
-        if isinstance(response, Mapping):
-            if self.app.mode in {
-                AppMode.ADVANCED_CHAT.value,
-                AppMode.COMPLETION.value,
-                AppMode.CHAT.value,
-                AppMode.AGENT_CHAT.value,
-            }:
-                answer = response["answer"]
-            elif self.app.mode in {AppMode.WORKFLOW.value}:
-                answer = json.dumps(response["data"]["outputs"], ensure_ascii=False)
-            else:
-                raise ValueError("Invalid app mode")
-            # Not support image yet
-        return types.CallToolResult(content=[types.TextContent(text=answer, type="text")])
-
-    def retrieve_end_user(self):
-        return (
-            db.session.query(EndUser)
-            .where(EndUser.external_user_id == self.mcp_server.id, EndUser.type == "mcp")
-            .first()
-        )
-
-    def _convert_input_form_to_parameters(self, user_input_form: list[VariableEntity]):
-        parameters: dict[str, dict[str, Any]] = {}
-        required = []
-        for item in user_input_form:
-            parameters[item.variable] = {}
-            if item.type in (
-                VariableEntityType.FILE,
-                VariableEntityType.FILE_LIST,
-                VariableEntityType.EXTERNAL_DATA_TOOL,
-            ):
-                continue
-            if item.required:
-                required.append(item.variable)
-            # if the workflow republished, the parameters not changed
-            # we should not raise error here
+def process_streaming_response(response: RateLimitGenerator) -> str:
+    """Process streaming response for agent chat mode"""
+    answer = ""
+    for item in response.generator:
+        if isinstance(item, str) and item.startswith("data: "):
             try:
-                description = self.mcp_server.parameters_dict[item.variable]
-            except KeyError:
-                description = ""
-            parameters[item.variable]["description"] = description
-            if item.type in (VariableEntityType.TEXT_INPUT, VariableEntityType.PARAGRAPH):
-                parameters[item.variable]["type"] = "string"
-            elif item.type == VariableEntityType.SELECT:
-                parameters[item.variable]["type"] = "string"
-                parameters[item.variable]["enum"] = item.options
-            elif item.type == VariableEntityType.NUMBER:
-                parameters[item.variable]["type"] = "float"
-        return parameters, required
+                json_str = item[6:].strip()
+                parsed_data = json.loads(json_str)
+                if parsed_data.get("event") == "agent_thought":
+                    answer += parsed_data.get("thought", "")
+            except json.JSONDecodeError:
+                continue
+    return answer
+
+
+def process_mapping_response(app: App, response: Mapping) -> str:
+    """Process mapping response based on app mode"""
+    if app.mode in {
+        AppMode.ADVANCED_CHAT.value,
+        AppMode.COMPLETION.value,
+        AppMode.CHAT.value,
+        AppMode.AGENT_CHAT.value,
+    }:
+        return response.get("answer", "")
+    elif app.mode == AppMode.WORKFLOW.value:
+        return json.dumps(response["data"]["outputs"], ensure_ascii=False)
+    else:
+        raise ValueError("Invalid app mode: " + str(app.mode))
+
+
+def convert_input_form_to_parameters(
+    user_input_form: list[VariableEntity],
+    parameters_dict: dict[str, str],
+) -> tuple[dict[str, dict[str, Any]], list[str]]:
+    """Convert user input form to parameter schema"""
+    parameters: dict[str, dict[str, Any]] = {}
+    required = []
+
+    for item in user_input_form:
+        if item.type in (
+            VariableEntityType.FILE,
+            VariableEntityType.FILE_LIST,
+            VariableEntityType.EXTERNAL_DATA_TOOL,
+        ):
+            continue
+        parameters[item.variable] = {}
+        if item.required:
+            required.append(item.variable)
+        # if the workflow republished, the parameters not changed
+        # we should not raise error here
+        description = parameters_dict.get(item.variable, "")
+        parameters[item.variable]["description"] = description
+        if item.type in (VariableEntityType.TEXT_INPUT, VariableEntityType.PARAGRAPH):
+            parameters[item.variable]["type"] = "string"
+        elif item.type == VariableEntityType.SELECT:
+            parameters[item.variable]["type"] = "string"
+            parameters[item.variable]["enum"] = item.options
+        elif item.type == VariableEntityType.NUMBER:
+            parameters[item.variable]["type"] = "float"
+    return parameters, required

api/core/mcp/utils.py

@@ -138,5 +138,5 @@ def create_mcp_error_response(
         error=error_data,
     )
     json_data = json.dumps(jsonable_encoder(json_response))
-    sse_content = f"event: message\ndata: {json_data}\n\n".encode()
+    sse_content = json_data.encode()
     yield sse_content

api/core/memory/token_buffer_memory.py

@@ -31,6 +31,65 @@ class TokenBufferMemory:
         self.conversation = conversation
         self.model_instance = model_instance
 
+    def _build_prompt_message_with_files(
+        self, message_files: list[MessageFile], text_content: str, message: Message, app_record, is_user_message: bool
+    ) -> PromptMessage:
+        """
+        Build prompt message with files.
+        :param message_files: list of MessageFile objects
+        :param text_content: text content of the message
+        :param message: Message object
+        :param app_record: app record
+        :param is_user_message: whether this is a user message
+        :return: PromptMessage
+        """
+        if self.conversation.mode in {AppMode.AGENT_CHAT, AppMode.COMPLETION, AppMode.CHAT}:
+            file_extra_config = FileUploadConfigManager.convert(self.conversation.model_config)
+        elif self.conversation.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
+            workflow_run = db.session.scalar(select(WorkflowRun).where(WorkflowRun.id == message.workflow_run_id))
+            if not workflow_run:
+                raise ValueError(f"Workflow run not found: {message.workflow_run_id}")
+            workflow = db.session.scalar(select(Workflow).where(Workflow.id == workflow_run.workflow_id))
+            if not workflow:
+                raise ValueError(f"Workflow not found: {workflow_run.workflow_id}")
+            file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
+        else:
+            raise AssertionError(f"Invalid app mode: {self.conversation.mode}")
+
+        detail = ImagePromptMessageContent.DETAIL.HIGH
+        if file_extra_config and app_record:
+            # Build files directly without filtering by belongs_to
+            file_objs = [
+                file_factory.build_from_message_file(
+                    message_file=message_file, tenant_id=app_record.tenant_id, config=file_extra_config
+                )
+                for message_file in message_files
+            ]
+            if file_extra_config.image_config and file_extra_config.image_config.detail:
+                detail = file_extra_config.image_config.detail
+        else:
+            file_objs = []
+
+        if not file_objs:
+            if is_user_message:
+                return UserPromptMessage(content=text_content)
+            else:
+                return AssistantPromptMessage(content=text_content)
+        else:
+            prompt_message_contents: list[PromptMessageContentUnionTypes] = []
+            for file in file_objs:
+                prompt_message = file_manager.to_prompt_message_content(
+                    file,
+                    image_detail_config=detail,
+                )
+                prompt_message_contents.append(prompt_message)
+            prompt_message_contents.append(TextPromptMessageContent(data=text_content))
+
+            if is_user_message:
+                return UserPromptMessage(content=prompt_message_contents)
+            else:
+                return AssistantPromptMessage(content=prompt_message_contents)
+
     def get_history_prompt_messages(
         self, max_token_limit: int = 2000, message_limit: Optional[int] = None
     ) -> Sequence[PromptMessage]:
@@ -51,9 +110,9 @@ class TokenBufferMemory:
         else:
             message_limit = 500
 
-        stmt = stmt.limit(message_limit)
+        msg_limit_stmt = stmt.limit(message_limit)
 
-        messages = db.session.scalars(stmt).all()
+        messages = db.session.scalars(msg_limit_stmt).all()
 
         # instead of all messages from the conversation, we only need to extract messages
         # that belong to the thread of last message
@@ -67,52 +126,46 @@ class TokenBufferMemory:
 
         prompt_messages: list[PromptMessage] = []
         for message in messages:
-            files = db.session.query(MessageFile).where(MessageFile.message_id == message.id).all()
-            if files:
-                file_extra_config = None
-                if self.conversation.mode in {AppMode.AGENT_CHAT, AppMode.COMPLETION, AppMode.CHAT}:
-                    file_extra_config = FileUploadConfigManager.convert(self.conversation.model_config)
-                elif self.conversation.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-                    workflow_run = db.session.scalar(
-                        select(WorkflowRun).where(WorkflowRun.id == message.workflow_run_id)
-                    )
-                    if not workflow_run:
-                        raise ValueError(f"Workflow run not found: {message.workflow_run_id}")
-                    workflow = db.session.scalar(select(Workflow).where(Workflow.id == workflow_run.workflow_id))
-                    if not workflow:
-                        raise ValueError(f"Workflow not found: {workflow_run.workflow_id}")
-                    file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
-                else:
-                    raise AssertionError(f"Invalid app mode: {self.conversation.mode}")
-
-                detail = ImagePromptMessageContent.DETAIL.LOW
-                if file_extra_config and app_record:
-                    file_objs = file_factory.build_from_message_files(
-                        message_files=files, tenant_id=app_record.tenant_id, config=file_extra_config
-                    )
-                    if file_extra_config.image_config and file_extra_config.image_config.detail:
-                        detail = file_extra_config.image_config.detail
-                else:
-                    file_objs = []
-
-                if not file_objs:
-                    prompt_messages.append(UserPromptMessage(content=message.query))
-                else:
-                    prompt_message_contents: list[PromptMessageContentUnionTypes] = []
-                    for file in file_objs:
-                        prompt_message = file_manager.to_prompt_message_content(
-                            file,
-                            image_detail_config=detail,
-                        )
-                        prompt_message_contents.append(prompt_message)
-                    prompt_message_contents.append(TextPromptMessageContent(data=message.query))
-
-                    prompt_messages.append(UserPromptMessage(content=prompt_message_contents))
+            # Process user message with files
+            user_files = (
+                db.session.query(MessageFile)
+                .where(
+                    MessageFile.message_id == message.id,
+                    (MessageFile.belongs_to == "user") | (MessageFile.belongs_to.is_(None)),
+                )
+                .all()
+            )
 
+            if user_files:
+                user_prompt_message = self._build_prompt_message_with_files(
+                    message_files=user_files,
+                    text_content=message.query,
+                    message=message,
+                    app_record=app_record,
+                    is_user_message=True,
+                )
+                prompt_messages.append(user_prompt_message)
             else:
                 prompt_messages.append(UserPromptMessage(content=message.query))
 
-            prompt_messages.append(AssistantPromptMessage(content=message.answer))
+            # Process assistant message with files
+            assistant_files = (
+                db.session.query(MessageFile)
+                .where(MessageFile.message_id == message.id, MessageFile.belongs_to == "assistant")
+                .all()
+            )
+
+            if assistant_files:
+                assistant_prompt_message = self._build_prompt_message_with_files(
+                    message_files=assistant_files,
+                    text_content=message.answer,
+                    message=message,
+                    app_record=app_record,
+                    is_user_message=False,
+                )
+                prompt_messages.append(assistant_prompt_message)
+            else:
+                prompt_messages.append(AssistantPromptMessage(content=message.answer))
 
         if not prompt_messages:
             return []

api/core/model_manager.py

@@ -158,8 +158,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, LargeLanguageModel):
             raise Exception("Model type instance is not LargeLanguageModel")
-
-        self.model_type_instance = cast(LargeLanguageModel, self.model_type_instance)
         return cast(
             Union[LLMResult, Generator],
             self._round_robin_invoke(
@@ -188,8 +186,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, LargeLanguageModel):
             raise Exception("Model type instance is not LargeLanguageModel")
-
-        self.model_type_instance = cast(LargeLanguageModel, self.model_type_instance)
         return cast(
             int,
             self._round_robin_invoke(
@@ -214,8 +210,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TextEmbeddingModel):
             raise Exception("Model type instance is not TextEmbeddingModel")
-
-        self.model_type_instance = cast(TextEmbeddingModel, self.model_type_instance)
         return cast(
             TextEmbeddingResult,
             self._round_robin_invoke(
@@ -237,8 +231,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TextEmbeddingModel):
             raise Exception("Model type instance is not TextEmbeddingModel")
-
-        self.model_type_instance = cast(TextEmbeddingModel, self.model_type_instance)
         return cast(
             list[int],
             self._round_robin_invoke(
@@ -269,8 +261,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, RerankModel):
             raise Exception("Model type instance is not RerankModel")
-
-        self.model_type_instance = cast(RerankModel, self.model_type_instance)
         return cast(
             RerankResult,
             self._round_robin_invoke(
@@ -295,8 +285,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, ModerationModel):
             raise Exception("Model type instance is not ModerationModel")
-
-        self.model_type_instance = cast(ModerationModel, self.model_type_instance)
         return cast(
             bool,
             self._round_robin_invoke(
@@ -318,8 +306,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, Speech2TextModel):
             raise Exception("Model type instance is not Speech2TextModel")
-
-        self.model_type_instance = cast(Speech2TextModel, self.model_type_instance)
         return cast(
             str,
             self._round_robin_invoke(
@@ -343,8 +329,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TTSModel):
             raise Exception("Model type instance is not TTSModel")
-
-        self.model_type_instance = cast(TTSModel, self.model_type_instance)
         return cast(
             Iterable[bytes],
             self._round_robin_invoke(
@@ -404,8 +388,6 @@ class ModelInstance:
         """
         if not isinstance(self.model_type_instance, TTSModel):
             raise Exception("Model type instance is not TTSModel")
-
-        self.model_type_instance = cast(TTSModel, self.model_type_instance)
         return self.model_type_instance.get_tts_model_voices(
             model=self.model, credentials=self.credentials, language=language
         )

api/core/model_runtime/entities/message_entities.py

@@ -87,6 +87,7 @@ class MultiModalPromptMessageContent(PromptMessageContent):
     base64_data: str = Field(default="", description="the base64 data of multi-modal file")
     url: str = Field(default="", description="the url of multi-modal file")
     mime_type: str = Field(default=..., description="the mime type of multi-modal file")
+    filename: str = Field(default="", description="the filename of multi-modal file")
 
     @property
     def data(self):

api/core/model_runtime/model_providers/__base/tokenizers/gpt2_tokenizer.py

@@ -43,7 +43,7 @@ class GPT2Tokenizer:
                 except Exception:
                     from os.path import abspath, dirname, join
 
-                    from transformers import GPT2Tokenizer as TransformerGPT2Tokenizer  # type: ignore
+                    from transformers import GPT2Tokenizer as TransformerGPT2Tokenizer
 
                     base_path = abspath(__file__)
                     gpt2_tokenizer_path = join(dirname(base_path), "gpt2")

api/core/moderation/api/api.py

@@ -1,6 +1,7 @@
 from typing import Optional
 
 from pydantic import BaseModel, Field
+from sqlalchemy import select
 
 from core.extension.api_based_extension_requestor import APIBasedExtensionPoint, APIBasedExtensionRequestor
 from core.helper.encrypter import decrypt_token
@@ -87,10 +88,9 @@ class ApiModeration(Moderation):
 
     @staticmethod
     def _get_api_based_extension(tenant_id: str, api_based_extension_id: str) -> Optional[APIBasedExtension]:
-        extension = (
-            db.session.query(APIBasedExtension)
-            .where(APIBasedExtension.tenant_id == tenant_id, APIBasedExtension.id == api_based_extension_id)
-            .first()
+        stmt = select(APIBasedExtension).where(
+            APIBasedExtension.tenant_id == tenant_id, APIBasedExtension.id == api_based_extension_id
         )
+        extension = db.session.scalar(stmt)
 
         return extension

api/core/moderation/factory.py

@@ -20,7 +20,6 @@ class ModerationFactory:
         :param config: the form config data
         :return:
         """
-        code_based_extension.validate_form_schema(ExtensionModule.MODERATION, name, config)
         extension_class = code_based_extension.extension_class(ExtensionModule.MODERATION, name)
         # FIXME: mypy error, try to fix it instead of using type: ignore
         extension_class.validate_config(tenant_id, config)  # type: ignore

api/core/moderation/output_moderation.py

@@ -135,7 +135,7 @@ class OutputModeration(BaseModel):
 
             result: ModerationOutputsResult = moderation_factory.moderation_for_outputs(moderation_buffer)
             return result
-        except Exception as e:
+        except Exception:
             logger.exception("Moderation Output error, app_id: %s", app_id)
 
         return None

api/core/ops/aliyun_trace/aliyun_trace.py

@@ -5,6 +5,7 @@ from typing import Optional
 from urllib.parse import urljoin
 
 from opentelemetry.trace import Link, Status, StatusCode
+from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.ops.aliyun_trace.data_exporter.traceclient import (
@@ -263,15 +264,15 @@ class AliyunDataTrace(BaseTraceInstance):
             app_id = trace_info.metadata.get("app_id")
             if not app_id:
                 raise ValueError("No app_id found in trace_info metadata")
-
-            app = session.query(App).where(App.id == app_id).first()
+            app_stmt = select(App).where(App.id == app_id)
+            app = session.scalar(app_stmt)
             if not app:
                 raise ValueError(f"App with id {app_id} not found")
 
             if not app.created_by:
                 raise ValueError(f"App with id {app_id} has no creator (created_by is None)")
-
-            service_account = session.query(Account).where(Account.id == app.created_by).first()
+            account_stmt = select(Account).where(Account.id == app.created_by)
+            service_account = session.scalar(account_stmt)
             if not service_account:
                 raise ValueError(f"Creator account with id {app.created_by} not found for app {app_id}")
             current_tenant = (

api/core/ops/aliyun_trace/data_exporter/traceclient.py

@@ -72,7 +72,7 @@ class TraceClient:
             else:
                 logger.debug("AliyunTrace API check failed: Unexpected status code: %s", response.status_code)
                 return False
-        except requests.exceptions.RequestException as e:
+        except requests.RequestException as e:
             logger.debug("AliyunTrace API check failed: %s", str(e))
             raise ValueError(f"AliyunTrace API check failed: {str(e)}")
 

api/core/ops/base_trace_instance.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
 
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from core.ops.entities.config_entity import BaseTracingConfig
@@ -44,14 +45,15 @@ class BaseTraceInstance(ABC):
         """
         with Session(db.engine, expire_on_commit=False) as session:
             # Get the app to find its creator
-            app = session.query(App).where(App.id == app_id).first()
+            app_stmt = select(App).where(App.id == app_id)
+            app = session.scalar(app_stmt)
             if not app:
                 raise ValueError(f"App with id {app_id} not found")
 
             if not app.created_by:
                 raise ValueError(f"App with id {app_id} has no creator (created_by is None)")
-
-            service_account = session.query(Account).where(Account.id == app.created_by).first()
+            account_stmt = select(Account).where(Account.id == app.created_by)
+            service_account = session.scalar(account_stmt)
             if not service_account:
                 raise ValueError(f"Creator account with id {app.created_by} not found for app {app_id}")