ECO-181: initial draft

- Added support for end-user credentials with options for OAuth and API Key.
- Introduced new props in PluginAuth for managing end-user credential types and their states.
- Updated workflow types to include end-user credential fields.
- Enhanced UI to allow users to select and manage end-user credentials.
- Added translations for new UI elements related to end-user credentials.

.github/CODEOWNERS

@@ -1,226 +0,0 @@
-# CODEOWNERS
-# This file defines code ownership for the Dify project.
-# Each line is a file pattern followed by one or more owners.
-# Owners can be @username, @org/team-name, or email addresses.
-# For more information, see: https://docs.github.com/en/repositories/managing-your-repositorys-settings-and-features/customizing-your-repository/about-code-owners
-
-* @crazywoola @laipz8200 @Yeuoly
-
-# Backend (default owner, more specific rules below will override)
-api/ @QuantumGhost
-
-# Backend - Workflow - Engine (Core graph execution engine)
-api/core/workflow/graph_engine/ @laipz8200 @QuantumGhost
-api/core/workflow/runtime/ @laipz8200 @QuantumGhost
-api/core/workflow/graph/ @laipz8200 @QuantumGhost
-api/core/workflow/graph_events/ @laipz8200 @QuantumGhost
-api/core/workflow/node_events/ @laipz8200 @QuantumGhost
-api/core/model_runtime/ @laipz8200 @QuantumGhost
-
-# Backend - Workflow - Nodes (Agent, Iteration, Loop, LLM)
-api/core/workflow/nodes/agent/ @Nov1c444
-api/core/workflow/nodes/iteration/ @Nov1c444
-api/core/workflow/nodes/loop/ @Nov1c444
-api/core/workflow/nodes/llm/ @Nov1c444
-
-# Backend - RAG (Retrieval Augmented Generation)
-api/core/rag/ @JohnJyong
-api/services/rag_pipeline/ @JohnJyong
-api/services/dataset_service.py @JohnJyong
-api/services/knowledge_service.py @JohnJyong
-api/services/external_knowledge_service.py @JohnJyong
-api/services/hit_testing_service.py @JohnJyong
-api/services/metadata_service.py @JohnJyong
-api/services/vector_service.py @JohnJyong
-api/services/entities/knowledge_entities/ @JohnJyong
-api/services/entities/external_knowledge_entities/ @JohnJyong
-api/controllers/console/datasets/ @JohnJyong
-api/controllers/service_api/dataset/ @JohnJyong
-api/models/dataset.py @JohnJyong
-api/tasks/rag_pipeline/ @JohnJyong
-api/tasks/add_document_to_index_task.py @JohnJyong
-api/tasks/batch_clean_document_task.py @JohnJyong
-api/tasks/clean_document_task.py @JohnJyong
-api/tasks/clean_notion_document_task.py @JohnJyong
-api/tasks/document_indexing_task.py @JohnJyong
-api/tasks/document_indexing_sync_task.py @JohnJyong
-api/tasks/document_indexing_update_task.py @JohnJyong
-api/tasks/duplicate_document_indexing_task.py @JohnJyong
-api/tasks/recover_document_indexing_task.py @JohnJyong
-api/tasks/remove_document_from_index_task.py @JohnJyong
-api/tasks/retry_document_indexing_task.py @JohnJyong
-api/tasks/sync_website_document_indexing_task.py @JohnJyong
-api/tasks/batch_create_segment_to_index_task.py @JohnJyong
-api/tasks/create_segment_to_index_task.py @JohnJyong
-api/tasks/delete_segment_from_index_task.py @JohnJyong
-api/tasks/disable_segment_from_index_task.py @JohnJyong
-api/tasks/disable_segments_from_index_task.py @JohnJyong
-api/tasks/enable_segment_to_index_task.py @JohnJyong
-api/tasks/enable_segments_to_index_task.py @JohnJyong
-api/tasks/clean_dataset_task.py @JohnJyong
-api/tasks/deal_dataset_index_update_task.py @JohnJyong
-api/tasks/deal_dataset_vector_index_task.py @JohnJyong
-
-# Backend - Plugins
-api/core/plugin/ @Mairuis @Yeuoly @Stream29
-api/services/plugin/ @Mairuis @Yeuoly @Stream29
-api/controllers/console/workspace/plugin.py @Mairuis @Yeuoly @Stream29
-api/controllers/inner_api/plugin/ @Mairuis @Yeuoly @Stream29
-api/tasks/process_tenant_plugin_autoupgrade_check_task.py @Mairuis @Yeuoly @Stream29
-
-# Backend - Trigger/Schedule/Webhook
-api/controllers/trigger/ @Mairuis @Yeuoly
-api/controllers/console/app/workflow_trigger.py @Mairuis @Yeuoly
-api/controllers/console/workspace/trigger_providers.py @Mairuis @Yeuoly
-api/core/trigger/ @Mairuis @Yeuoly
-api/core/app/layers/trigger_post_layer.py @Mairuis @Yeuoly
-api/services/trigger/ @Mairuis @Yeuoly
-api/models/trigger.py @Mairuis @Yeuoly
-api/fields/workflow_trigger_fields.py @Mairuis @Yeuoly
-api/repositories/workflow_trigger_log_repository.py @Mairuis @Yeuoly
-api/repositories/sqlalchemy_workflow_trigger_log_repository.py @Mairuis @Yeuoly
-api/libs/schedule_utils.py @Mairuis @Yeuoly
-api/services/workflow/scheduler.py @Mairuis @Yeuoly
-api/schedule/trigger_provider_refresh_task.py @Mairuis @Yeuoly
-api/schedule/workflow_schedule_task.py @Mairuis @Yeuoly
-api/tasks/trigger_processing_tasks.py @Mairuis @Yeuoly
-api/tasks/trigger_subscription_refresh_tasks.py @Mairuis @Yeuoly
-api/tasks/workflow_schedule_tasks.py @Mairuis @Yeuoly
-api/tasks/workflow_cfs_scheduler/ @Mairuis @Yeuoly
-api/events/event_handlers/sync_plugin_trigger_when_app_created.py @Mairuis @Yeuoly
-api/events/event_handlers/update_app_triggers_when_app_published_workflow_updated.py @Mairuis @Yeuoly
-api/events/event_handlers/sync_workflow_schedule_when_app_published.py @Mairuis @Yeuoly
-api/events/event_handlers/sync_webhook_when_app_created.py @Mairuis @Yeuoly
-
-# Backend - Async Workflow
-api/services/async_workflow_service.py @Mairuis @Yeuoly
-api/tasks/async_workflow_tasks.py @Mairuis @Yeuoly
-
-# Backend - Billing
-api/services/billing_service.py @hj24 @zyssyz123
-api/controllers/console/billing/ @hj24 @zyssyz123
-
-# Backend - Enterprise
-api/configs/enterprise/ @GarfieldDai @GareArc
-api/services/enterprise/ @GarfieldDai @GareArc
-api/services/feature_service.py @GarfieldDai @GareArc
-api/controllers/console/feature.py @GarfieldDai @GareArc
-api/controllers/web/feature.py @GarfieldDai @GareArc
-
-# Backend - Database Migrations
-api/migrations/ @snakevash @laipz8200
-
-# Frontend
-web/ @iamjoel
-
-# Frontend - App - Orchestration
-web/app/components/workflow/ @iamjoel @zxhlyh
-web/app/components/workflow-app/ @iamjoel @zxhlyh
-web/app/components/app/configuration/ @iamjoel @zxhlyh
-web/app/components/app/app-publisher/ @iamjoel @zxhlyh
-
-# Frontend - WebApp - Chat
-web/app/components/base/chat/ @iamjoel @zxhlyh
-
-# Frontend - WebApp - Completion
-web/app/components/share/text-generation/ @iamjoel @zxhlyh
-
-# Frontend - App - List and Creation
-web/app/components/apps/ @JzoNgKVO @iamjoel
-web/app/components/app/create-app-dialog/ @JzoNgKVO @iamjoel
-web/app/components/app/create-app-modal/ @JzoNgKVO @iamjoel
-web/app/components/app/create-from-dsl-modal/ @JzoNgKVO @iamjoel
-
-# Frontend - App - API Documentation
-web/app/components/develop/ @JzoNgKVO @iamjoel
-
-# Frontend - App - Logs and Annotations
-web/app/components/app/workflow-log/ @JzoNgKVO @iamjoel
-web/app/components/app/log/ @JzoNgKVO @iamjoel
-web/app/components/app/log-annotation/ @JzoNgKVO @iamjoel
-web/app/components/app/annotation/ @JzoNgKVO @iamjoel
-
-# Frontend - App - Monitoring
-web/app/(commonLayout)/app/(appDetailLayout)/\[appId\]/overview/ @JzoNgKVO @iamjoel
-web/app/components/app/overview/ @JzoNgKVO @iamjoel
-
-# Frontend - App - Settings
-web/app/components/app-sidebar/ @JzoNgKVO @iamjoel
-
-# Frontend - RAG - Hit Testing
-web/app/components/datasets/hit-testing/ @JzoNgKVO @iamjoel
-
-# Frontend - RAG - List and Creation
-web/app/components/datasets/list/ @iamjoel @WTW0313
-web/app/components/datasets/create/ @iamjoel @WTW0313
-web/app/components/datasets/create-from-pipeline/ @iamjoel @WTW0313
-web/app/components/datasets/external-knowledge-base/ @iamjoel @WTW0313
-
-# Frontend - RAG - Orchestration (general rule first, specific rules below override)
-web/app/components/rag-pipeline/ @iamjoel @WTW0313
-web/app/components/rag-pipeline/components/rag-pipeline-main.tsx @iamjoel @zxhlyh
-web/app/components/rag-pipeline/store/ @iamjoel @zxhlyh
-
-# Frontend - RAG - Documents List
-web/app/components/datasets/documents/list.tsx @iamjoel @WTW0313
-web/app/components/datasets/documents/create-from-pipeline/ @iamjoel @WTW0313
-
-# Frontend - RAG - Segments List
-web/app/components/datasets/documents/detail/ @iamjoel @WTW0313
-
-# Frontend - RAG - Settings
-web/app/components/datasets/settings/ @iamjoel @WTW0313
-
-# Frontend - Ecosystem - Plugins
-web/app/components/plugins/ @iamjoel @zhsama
-
-# Frontend - Ecosystem - Tools
-web/app/components/tools/ @iamjoel @Yessenia-d
-
-# Frontend - Ecosystem - MarketPlace
-web/app/components/plugins/marketplace/ @iamjoel @Yessenia-d
-
-# Frontend - Login and Registration
-web/app/signin/ @douxc @iamjoel
-web/app/signup/ @douxc @iamjoel
-web/app/reset-password/ @douxc @iamjoel
-web/app/install/ @douxc @iamjoel
-web/app/init/ @douxc @iamjoel
-web/app/forgot-password/ @douxc @iamjoel
-web/app/account/ @douxc @iamjoel
-
-# Frontend - Service Authentication
-web/service/base.ts @douxc @iamjoel
-
-# Frontend - WebApp Authentication and Access Control
-web/app/(shareLayout)/components/ @douxc @iamjoel
-web/app/(shareLayout)/webapp-signin/ @douxc @iamjoel
-web/app/(shareLayout)/webapp-reset-password/ @douxc @iamjoel
-web/app/components/app/app-access-control/ @douxc @iamjoel
-
-# Frontend - Explore Page
-web/app/components/explore/ @CodingOnStar @iamjoel
-
-# Frontend - Personal Settings
-web/app/components/header/account-setting/ @CodingOnStar @iamjoel
-web/app/components/header/account-dropdown/ @CodingOnStar @iamjoel
-
-# Frontend - Analytics
-web/app/components/base/ga/ @CodingOnStar @iamjoel
-
-# Frontend - Base Components
-web/app/components/base/ @iamjoel @zxhlyh
-
-# Frontend - Utils and Hooks
-web/utils/classnames.ts @iamjoel @zxhlyh
-web/utils/time.ts @iamjoel @zxhlyh
-web/utils/format.ts @iamjoel @zxhlyh
-web/utils/clipboard.ts @iamjoel @zxhlyh
-web/hooks/use-document-title.ts @iamjoel @zxhlyh
-
-# Frontend - Billing and Education
-web/app/components/billing/ @iamjoel @zxhlyh
-web/app/education-apply/ @iamjoel @zxhlyh
-
-# Frontend - Workspace
-web/app/components/header/account-dropdown/workplace-selector/ @iamjoel @zxhlyh

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

@@ -77,15 +77,12 @@ jobs:
         uses: peter-evans/create-pull-request@v6
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
-          commit-message: 'chore(i18n): update translations based on en-US changes'
-          title: 'chore(i18n): translate i18n files and update type definitions'
+          commit-message: Update i18n files and type definitions based on en-US changes
+          title: 'chore: translate i18n files and update type definitions'
           body: |
             This PR was automatically created to update i18n files and TypeScript type definitions based on changes in en-US locale.
-
-            **Triggered by:** ${{ github.sha }}
-
+            
             **Changes included:**
             - Updated translation files for all locales
             - Regenerated TypeScript type definitions for type safety
-          branch: chore/automated-i18n-updates-${{ github.sha }}
-          delete-branch: true
+          branch: chore/automated-i18n-updates

api/Dockerfile

@@ -48,12 +48,6 @@ ENV PYTHONIOENCODING=utf-8
 
 WORKDIR /app/api
 
-# Create non-root user
-ARG dify_uid=1001
-RUN groupadd -r -g ${dify_uid} dify && \
-    useradd -r -u ${dify_uid} -g ${dify_uid} -s /bin/bash dify && \
-    chown -R dify:dify /app
-
 RUN \
     apt-get update \
     # Install dependencies
@@ -75,7 +69,7 @@ RUN \
 
 # Copy Python environment and packages
 ENV VIRTUAL_ENV=/app/api/.venv
-COPY --from=packages --chown=dify:dify ${VIRTUAL_ENV} ${VIRTUAL_ENV}
+COPY --from=packages ${VIRTUAL_ENV} ${VIRTUAL_ENV}
 ENV PATH="${VIRTUAL_ENV}/bin:${PATH}"
 
 # Download nltk data
@@ -84,20 +78,24 @@ RUN mkdir -p /usr/local/share/nltk_data && NLTK_DATA=/usr/local/share/nltk_data
 
 ENV TIKTOKEN_CACHE_DIR=/app/api/.tiktoken_cache
 
-RUN python -c "import tiktoken; tiktoken.encoding_for_model('gpt2')" \
-    && chown -R dify:dify ${TIKTOKEN_CACHE_DIR}
+RUN python -c "import tiktoken; tiktoken.encoding_for_model('gpt2')"
 
 # Copy source code
-COPY --chown=dify:dify . /app/api/
+COPY . /app/api/
 
-# Prepare entrypoint script
-COPY --chown=dify:dify --chmod=755 docker/entrypoint.sh /entrypoint.sh
+# Copy entrypoint
+COPY docker/entrypoint.sh /entrypoint.sh
+RUN chmod +x /entrypoint.sh
 
+# Create non-root user and set permissions
+RUN groupadd -r -g 1001 dify && \
+    useradd -r -u 1001 -g 1001 -s /bin/bash dify && \
+    mkdir -p /home/dify && \
+    chown -R 1001:1001 /app /home/dify ${TIKTOKEN_CACHE_DIR} /entrypoint.sh
 
 ARG COMMIT_SHA
 ENV COMMIT_SHA=${COMMIT_SHA}
 ENV NLTK_DATA=/usr/local/share/nltk_data
-
-USER dify
+USER 1001
 
 ENTRYPOINT ["/bin/bash", "/entrypoint.sh"]

api/app_factory.py

@@ -51,7 +51,6 @@ def initialize_extensions(app: DifyApp):
         ext_commands,
         ext_compress,
         ext_database,
-        ext_forward_refs,
         ext_hosting_provider,
         ext_import_modules,
         ext_logging,
@@ -76,7 +75,6 @@ def initialize_extensions(app: DifyApp):
         ext_warnings,
         ext_import_modules,
         ext_orjson,
-        ext_forward_refs,
         ext_set_secretkey,
         ext_compress,
         ext_code_based_extension,

api/controllers/console/app/advanced_prompt_template.py

@@ -1,23 +1,16 @@
-from flask import request
-from flask_restx import Resource, fields
-from pydantic import BaseModel, Field
+from flask_restx import Resource, fields, reqparse
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, setup_required
 from libs.login import login_required
 from services.advanced_prompt_template_service import AdvancedPromptTemplateService
 
-
-class AdvancedPromptTemplateQuery(BaseModel):
-    app_mode: str = Field(..., description="Application mode")
-    model_mode: str = Field(..., description="Model mode")
-    has_context: str = Field(default="true", description="Whether has context")
-    model_name: str = Field(..., description="Model name")
-
-
-console_ns.schema_model(
-    AdvancedPromptTemplateQuery.__name__,
-    AdvancedPromptTemplateQuery.model_json_schema(ref_template="#/definitions/{model}"),
+parser = (
+    reqparse.RequestParser()
+    .add_argument("app_mode", type=str, required=True, location="args", help="Application mode")
+    .add_argument("model_mode", type=str, required=True, location="args", help="Model mode")
+    .add_argument("has_context", type=str, required=False, default="true", location="args", help="Whether has context")
+    .add_argument("model_name", type=str, required=True, location="args", help="Model name")
 )
 
 
@@ -25,7 +18,7 @@ console_ns.schema_model(
 class AdvancedPromptTemplateList(Resource):
     @console_ns.doc("get_advanced_prompt_templates")
     @console_ns.doc(description="Get advanced prompt templates based on app mode and model configuration")
-    @console_ns.expect(console_ns.models[AdvancedPromptTemplateQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200, "Prompt templates retrieved successfully", fields.List(fields.Raw(description="Prompt template data"))
     )
@@ -34,6 +27,6 @@ class AdvancedPromptTemplateList(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        args = AdvancedPromptTemplateQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
-        return AdvancedPromptTemplateService.get_prompt(args.model_dump())
+        return AdvancedPromptTemplateService.get_prompt(args)

api/controllers/console/app/app.py

@@ -1,12 +1,9 @@
 import uuid
-from typing import Literal
 
-from flask import request
-from flask_restx import Resource, fields, marshal, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource, fields, inputs, marshal, marshal_with, reqparse
 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import BadRequest
+from werkzeug.exceptions import BadRequest, abort
 
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
@@ -39,130 +36,6 @@ from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
 ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "completion"]
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class AppListQuery(BaseModel):
-    page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
-    limit: int = Field(default=20, ge=1, le=100, description="Page size (1-100)")
-    mode: Literal["completion", "chat", "advanced-chat", "workflow", "agent-chat", "channel", "all"] = Field(
-        default="all", description="App mode filter"
-    )
-    name: str | None = Field(default=None, description="Filter by app name")
-    tag_ids: list[str] | None = Field(default=None, description="Comma-separated tag IDs")
-    is_created_by_me: bool | None = Field(default=None, description="Filter by creator")
-
-    @field_validator("tag_ids", mode="before")
-    @classmethod
-    def validate_tag_ids(cls, value: str | list[str] | None) -> list[str] | None:
-        if not value:
-            return None
-
-        if isinstance(value, str):
-            items = [item.strip() for item in value.split(",") if item.strip()]
-        elif isinstance(value, list):
-            items = [str(item).strip() for item in value if item and str(item).strip()]
-        else:
-            raise TypeError("Unsupported tag_ids type.")
-
-        if not items:
-            return None
-
-        try:
-            return [str(uuid.UUID(item)) for item in items]
-        except ValueError as exc:
-            raise ValueError("Invalid UUID format in tag_ids.") from exc
-
-
-class CreateAppPayload(BaseModel):
-    name: str = Field(..., min_length=1, description="App name")
-    description: str | None = Field(default=None, description="App description (max 400 chars)")
-    mode: Literal["chat", "agent-chat", "advanced-chat", "workflow", "completion"] = Field(..., description="App mode")
-    icon_type: str | None = Field(default=None, description="Icon type")
-    icon: str | None = Field(default=None, description="Icon")
-    icon_background: str | None = Field(default=None, description="Icon background color")
-
-    @field_validator("description")
-    @classmethod
-    def validate_description(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return validate_description_length(value)
-
-
-class UpdateAppPayload(BaseModel):
-    name: str = Field(..., min_length=1, description="App name")
-    description: str | None = Field(default=None, description="App description (max 400 chars)")
-    icon_type: str | None = Field(default=None, description="Icon type")
-    icon: str | None = Field(default=None, description="Icon")
-    icon_background: str | None = Field(default=None, description="Icon background color")
-    use_icon_as_answer_icon: bool | None = Field(default=None, description="Use icon as answer icon")
-    max_active_requests: int | None = Field(default=None, description="Maximum active requests")
-
-    @field_validator("description")
-    @classmethod
-    def validate_description(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return validate_description_length(value)
-
-
-class CopyAppPayload(BaseModel):
-    name: str | None = Field(default=None, description="Name for the copied app")
-    description: str | None = Field(default=None, description="Description for the copied app")
-    icon_type: str | None = Field(default=None, description="Icon type")
-    icon: str | None = Field(default=None, description="Icon")
-    icon_background: str | None = Field(default=None, description="Icon background color")
-
-    @field_validator("description")
-    @classmethod
-    def validate_description(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return validate_description_length(value)
-
-
-class AppExportQuery(BaseModel):
-    include_secret: bool = Field(default=False, description="Include secrets in export")
-    workflow_id: str | None = Field(default=None, description="Specific workflow ID to export")
-
-
-class AppNamePayload(BaseModel):
-    name: str = Field(..., min_length=1, description="Name to check")
-
-
-class AppIconPayload(BaseModel):
-    icon: str | None = Field(default=None, description="Icon data")
-    icon_background: str | None = Field(default=None, description="Icon background color")
-
-
-class AppSiteStatusPayload(BaseModel):
-    enable_site: bool = Field(..., description="Enable or disable site")
-
-
-class AppApiStatusPayload(BaseModel):
-    enable_api: bool = Field(..., description="Enable or disable API")
-
-
-class AppTracePayload(BaseModel):
-    enabled: bool = Field(..., description="Enable or disable tracing")
-    tracing_provider: str = Field(..., description="Tracing provider")
-
-
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(AppListQuery)
-reg(CreateAppPayload)
-reg(UpdateAppPayload)
-reg(CopyAppPayload)
-reg(AppExportQuery)
-reg(AppNamePayload)
-reg(AppIconPayload)
-reg(AppSiteStatusPayload)
-reg(AppApiStatusPayload)
-reg(AppTracePayload)
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register base models first
@@ -274,7 +147,22 @@ app_pagination_model = console_ns.model(
 class AppListApi(Resource):
     @console_ns.doc("list_apps")
     @console_ns.doc(description="Get list of applications with pagination and filtering")
-    @console_ns.expect(console_ns.models[AppListQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("page", type=int, location="args", help="Page number (1-99999)", default=1)
+        .add_argument("limit", type=int, location="args", help="Page size (1-100)", default=20)
+        .add_argument(
+            "mode",
+            type=str,
+            location="args",
+            choices=["completion", "chat", "advanced-chat", "workflow", "agent-chat", "channel", "all"],
+            default="all",
+            help="App mode filter",
+        )
+        .add_argument("name", type=str, location="args", help="Filter by app name")
+        .add_argument("tag_ids", type=str, location="args", help="Comma-separated tag IDs")
+        .add_argument("is_created_by_me", type=bool, location="args", help="Filter by creator")
+    )
     @console_ns.response(200, "Success", app_pagination_model)
     @setup_required
     @login_required
@@ -284,12 +172,42 @@ class AppListApi(Resource):
         """Get app list"""
         current_user, current_tenant_id = current_account_with_tenant()
 
-        args = AppListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        args_dict = args.model_dump()
+        def uuid_list(value):
+            try:
+                return [str(uuid.UUID(v)) for v in value.split(",")]
+            except ValueError:
+                abort(400, message="Invalid UUID format in tag_ids.")
+
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("page", type=inputs.int_range(1, 99999), required=False, default=1, location="args")
+            .add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args")
+            .add_argument(
+                "mode",
+                type=str,
+                choices=[
+                    "completion",
+                    "chat",
+                    "advanced-chat",
+                    "workflow",
+                    "agent-chat",
+                    "channel",
+                    "all",
+                ],
+                default="all",
+                location="args",
+                required=False,
+            )
+            .add_argument("name", type=str, location="args", required=False)
+            .add_argument("tag_ids", type=uuid_list, location="args", required=False)
+            .add_argument("is_created_by_me", type=inputs.boolean, location="args", required=False)
+        )
+
+        args = parser.parse_args()
 
         # get app list
         app_service = AppService()
-        app_pagination = app_service.get_paginate_apps(current_user.id, current_tenant_id, args_dict)
+        app_pagination = app_service.get_paginate_apps(current_user.id, current_tenant_id, args)
         if not app_pagination:
             return {"data": [], "total": 0, "page": 1, "limit": 20, "has_more": False}
 
@@ -336,7 +254,19 @@ class AppListApi(Resource):
 
     @console_ns.doc("create_app")
     @console_ns.doc(description="Create a new application")
-    @console_ns.expect(console_ns.models[CreateAppPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "CreateAppRequest",
+            {
+                "name": fields.String(required=True, description="App name"),
+                "description": fields.String(description="App description (max 400 chars)"),
+                "mode": fields.String(required=True, enum=ALLOW_CREATE_APP_MODES, description="App mode"),
+                "icon_type": fields.String(description="Icon type"),
+                "icon": fields.String(description="Icon"),
+                "icon_background": fields.String(description="Icon background color"),
+            },
+        )
+    )
     @console_ns.response(201, "App created successfully", app_detail_model)
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(400, "Invalid request parameters")
@@ -349,10 +279,22 @@ class AppListApi(Resource):
     def post(self):
         """Create app"""
         current_user, current_tenant_id = current_account_with_tenant()
-        args = CreateAppPayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("name", type=str, required=True, location="json")
+            .add_argument("description", type=validate_description_length, location="json")
+            .add_argument("mode", type=str, choices=ALLOW_CREATE_APP_MODES, location="json")
+            .add_argument("icon_type", type=str, location="json")
+            .add_argument("icon", type=str, location="json")
+            .add_argument("icon_background", type=str, location="json")
+        )
+        args = parser.parse_args()
+
+        if "mode" not in args or args["mode"] is None:
+            raise BadRequest("mode is required")
 
         app_service = AppService()
-        app = app_service.create_app(current_tenant_id, args.model_dump(), current_user)
+        app = app_service.create_app(current_tenant_id, args, current_user)
 
         return app, 201
 
@@ -384,7 +326,20 @@ class AppApi(Resource):
     @console_ns.doc("update_app")
     @console_ns.doc(description="Update application details")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[UpdateAppPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "UpdateAppRequest",
+            {
+                "name": fields.String(required=True, description="App name"),
+                "description": fields.String(description="App description (max 400 chars)"),
+                "icon_type": fields.String(description="Icon type"),
+                "icon": fields.String(description="Icon"),
+                "icon_background": fields.String(description="Icon background color"),
+                "use_icon_as_answer_icon": fields.Boolean(description="Use icon as answer icon"),
+                "max_active_requests": fields.Integer(description="Maximum active requests"),
+            },
+        )
+    )
     @console_ns.response(200, "App updated successfully", app_detail_with_site_model)
     @console_ns.response(403, "Insufficient permissions")
     @console_ns.response(400, "Invalid request parameters")
@@ -396,18 +351,28 @@ class AppApi(Resource):
     @marshal_with(app_detail_with_site_model)
     def put(self, app_model):
         """Update app"""
-        args = UpdateAppPayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("name", type=str, required=True, nullable=False, location="json")
+            .add_argument("description", type=validate_description_length, location="json")
+            .add_argument("icon_type", type=str, location="json")
+            .add_argument("icon", type=str, location="json")
+            .add_argument("icon_background", type=str, location="json")
+            .add_argument("use_icon_as_answer_icon", type=bool, location="json")
+            .add_argument("max_active_requests", type=int, location="json")
+        )
+        args = parser.parse_args()
 
         app_service = AppService()
 
         args_dict: AppService.ArgsDict = {
-            "name": args.name,
-            "description": args.description or "",
-            "icon_type": args.icon_type or "",
-            "icon": args.icon or "",
-            "icon_background": args.icon_background or "",
-            "use_icon_as_answer_icon": args.use_icon_as_answer_icon or False,
-            "max_active_requests": args.max_active_requests or 0,
+            "name": args["name"],
+            "description": args.get("description", ""),
+            "icon_type": args.get("icon_type", ""),
+            "icon": args.get("icon", ""),
+            "icon_background": args.get("icon_background", ""),
+            "use_icon_as_answer_icon": args.get("use_icon_as_answer_icon", False),
+            "max_active_requests": args.get("max_active_requests", 0),
         }
         app_model = app_service.update_app(app_model, args_dict)
 
@@ -436,7 +401,18 @@ class AppCopyApi(Resource):
     @console_ns.doc("copy_app")
     @console_ns.doc(description="Create a copy of an existing application")
     @console_ns.doc(params={"app_id": "Application ID to copy"})
-    @console_ns.expect(console_ns.models[CopyAppPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "CopyAppRequest",
+            {
+                "name": fields.String(description="Name for the copied app"),
+                "description": fields.String(description="Description for the copied app"),
+                "icon_type": fields.String(description="Icon type"),
+                "icon": fields.String(description="Icon"),
+                "icon_background": fields.String(description="Icon background color"),
+            },
+        )
+    )
     @console_ns.response(201, "App copied successfully", app_detail_with_site_model)
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -450,7 +426,15 @@ class AppCopyApi(Resource):
         # The role of the current user in the ta table must be admin, owner, or editor
         current_user, _ = current_account_with_tenant()
 
-        args = CopyAppPayload.model_validate(console_ns.payload or {})
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("name", type=str, location="json")
+            .add_argument("description", type=validate_description_length, location="json")
+            .add_argument("icon_type", type=str, location="json")
+            .add_argument("icon", type=str, location="json")
+            .add_argument("icon_background", type=str, location="json")
+        )
+        args = parser.parse_args()
 
         with Session(db.engine) as session:
             import_service = AppDslService(session)
@@ -459,11 +443,11 @@ class AppCopyApi(Resource):
                 account=current_user,
                 import_mode=ImportMode.YAML_CONTENT,
                 yaml_content=yaml_content,
-                name=args.name,
-                description=args.description,
-                icon_type=args.icon_type,
-                icon=args.icon,
-                icon_background=args.icon_background,
+                name=args.get("name"),
+                description=args.get("description"),
+                icon_type=args.get("icon_type"),
+                icon=args.get("icon"),
+                icon_background=args.get("icon_background"),
             )
             session.commit()
 
@@ -478,7 +462,11 @@ class AppExportApi(Resource):
     @console_ns.doc("export_app")
     @console_ns.doc(description="Export application configuration as DSL")
     @console_ns.doc(params={"app_id": "Application ID to export"})
-    @console_ns.expect(console_ns.models[AppExportQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("include_secret", type=bool, location="args", default=False, help="Include secrets in export")
+        .add_argument("workflow_id", type=str, location="args", help="Specific workflow ID to export")
+    )
     @console_ns.response(
         200,
         "App exported successfully",
@@ -492,23 +480,30 @@ class AppExportApi(Resource):
     @edit_permission_required
     def get(self, app_model):
         """Export app"""
-        args = AppExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        # Add include_secret params
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("include_secret", type=inputs.boolean, default=False, location="args")
+            .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,
-                workflow_id=args.workflow_id,
+                app_model=app_model, include_secret=args["include_secret"], workflow_id=args.get("workflow_id")
             )
         }
 
 
+parser = reqparse.RequestParser().add_argument("name", type=str, required=True, location="json", help="Name to check")
+
+
 @console_ns.route("/apps/<uuid:app_id>/name")
 class AppNameApi(Resource):
     @console_ns.doc("check_app_name")
     @console_ns.doc(description="Check if app name is available")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[AppNamePayload.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(200, "Name availability checked")
     @setup_required
     @login_required
@@ -517,10 +512,10 @@ class AppNameApi(Resource):
     @marshal_with(app_detail_model)
     @edit_permission_required
     def post(self, app_model):
-        args = AppNamePayload.model_validate(console_ns.payload)
+        args = parser.parse_args()
 
         app_service = AppService()
-        app_model = app_service.update_app_name(app_model, args.name)
+        app_model = app_service.update_app_name(app_model, args["name"])
 
         return app_model
 
@@ -530,7 +525,16 @@ class AppIconApi(Resource):
     @console_ns.doc("update_app_icon")
     @console_ns.doc(description="Update application icon")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[AppIconPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "AppIconRequest",
+            {
+                "icon": fields.String(required=True, description="Icon data"),
+                "icon_type": fields.String(description="Icon type"),
+                "icon_background": fields.String(description="Icon background color"),
+            },
+        )
+    )
     @console_ns.response(200, "Icon updated successfully")
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -540,10 +544,15 @@ class AppIconApi(Resource):
     @marshal_with(app_detail_model)
     @edit_permission_required
     def post(self, app_model):
-        args = AppIconPayload.model_validate(console_ns.payload or {})
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("icon", type=str, location="json")
+            .add_argument("icon_background", type=str, location="json")
+        )
+        args = parser.parse_args()
 
         app_service = AppService()
-        app_model = app_service.update_app_icon(app_model, args.icon or "", args.icon_background or "")
+        app_model = app_service.update_app_icon(app_model, args.get("icon") or "", args.get("icon_background") or "")
 
         return app_model
 
@@ -553,7 +562,11 @@ class AppSiteStatus(Resource):
     @console_ns.doc("update_app_site_status")
     @console_ns.doc(description="Enable or disable app site")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[AppSiteStatusPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "AppSiteStatusRequest", {"enable_site": fields.Boolean(required=True, description="Enable or disable site")}
+        )
+    )
     @console_ns.response(200, "Site status updated successfully", app_detail_model)
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -563,10 +576,11 @@ class AppSiteStatus(Resource):
     @marshal_with(app_detail_model)
     @edit_permission_required
     def post(self, app_model):
-        args = AppSiteStatusPayload.model_validate(console_ns.payload)
+        parser = reqparse.RequestParser().add_argument("enable_site", type=bool, required=True, location="json")
+        args = parser.parse_args()
 
         app_service = AppService()
-        app_model = app_service.update_app_site_status(app_model, args.enable_site)
+        app_model = app_service.update_app_site_status(app_model, args["enable_site"])
 
         return app_model
 
@@ -576,7 +590,11 @@ class AppApiStatus(Resource):
     @console_ns.doc("update_app_api_status")
     @console_ns.doc(description="Enable or disable app API")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[AppApiStatusPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "AppApiStatusRequest", {"enable_api": fields.Boolean(required=True, description="Enable or disable API")}
+        )
+    )
     @console_ns.response(200, "API status updated successfully", app_detail_model)
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -586,10 +604,11 @@ class AppApiStatus(Resource):
     @get_app_model
     @marshal_with(app_detail_model)
     def post(self, app_model):
-        args = AppApiStatusPayload.model_validate(console_ns.payload)
+        parser = reqparse.RequestParser().add_argument("enable_api", type=bool, required=True, location="json")
+        args = parser.parse_args()
 
         app_service = AppService()
-        app_model = app_service.update_app_api_status(app_model, args.enable_api)
+        app_model = app_service.update_app_api_status(app_model, args["enable_api"])
 
         return app_model
 
@@ -612,7 +631,15 @@ class AppTraceApi(Resource):
     @console_ns.doc("update_app_trace")
     @console_ns.doc(description="Update app tracing configuration")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[AppTracePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "AppTraceRequest",
+            {
+                "enabled": fields.Boolean(required=True, description="Enable or disable tracing"),
+                "tracing_provider": fields.String(required=True, description="Tracing provider"),
+            },
+        )
+    )
     @console_ns.response(200, "Trace configuration updated successfully")
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -621,12 +648,17 @@ class AppTraceApi(Resource):
     @edit_permission_required
     def post(self, app_id):
         # add app trace
-        args = AppTracePayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("enabled", type=bool, required=True, location="json")
+            .add_argument("tracing_provider", type=str, required=True, location="json")
+        )
+        args = parser.parse_args()
 
         OpsTraceManager.update_app_tracing_config(
             app_id=app_id,
-            enabled=args.enabled,
-            tracing_provider=args.tracing_provider,
+            enabled=args["enabled"],
+            tracing_provider=args["tracing_provider"],
         )
 
         return {"result": "success"}

api/controllers/console/app/completion.py

@@ -1,9 +1,7 @@
 import logging
-from typing import Any, Literal
 
 from flask import request
-from flask_restx import Resource
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource, fields, reqparse
 from werkzeug.exceptions import InternalServerError, NotFound
 
 import services
@@ -37,41 +35,6 @@ from services.app_task_service import AppTaskService
 from services.errors.llm import InvokeRateLimitError
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class BaseMessagePayload(BaseModel):
-    inputs: dict[str, Any]
-    model_config_data: dict[str, Any] = Field(..., alias="model_config")
-    files: list[Any] | None = Field(default=None, description="Uploaded files")
-    response_mode: Literal["blocking", "streaming"] = Field(default="blocking", description="Response mode")
-    retriever_from: str = Field(default="dev", description="Retriever source")
-
-
-class CompletionMessagePayload(BaseMessagePayload):
-    query: str = Field(default="", description="Query text")
-
-
-class ChatMessagePayload(BaseMessagePayload):
-    query: str = Field(..., description="User query")
-    conversation_id: str | None = Field(default=None, description="Conversation ID")
-    parent_message_id: str | None = Field(default=None, description="Parent message ID")
-
-    @field_validator("conversation_id", "parent_message_id")
-    @classmethod
-    def validate_uuid(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return uuid_value(value)
-
-
-console_ns.schema_model(
-    CompletionMessagePayload.__name__,
-    CompletionMessagePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    ChatMessagePayload.__name__, ChatMessagePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
 
 
 # define completion message api for user
@@ -80,7 +43,19 @@ class CompletionMessageApi(Resource):
     @console_ns.doc("create_completion_message")
     @console_ns.doc(description="Generate completion message for debugging")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[CompletionMessagePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "CompletionMessageRequest",
+            {
+                "inputs": fields.Raw(required=True, description="Input variables"),
+                "query": fields.String(description="Query text", default=""),
+                "files": fields.List(fields.Raw(), description="Uploaded files"),
+                "model_config": fields.Raw(required=True, description="Model configuration"),
+                "response_mode": fields.String(enum=["blocking", "streaming"], description="Response mode"),
+                "retriever_from": fields.String(default="dev", description="Retriever source"),
+            },
+        )
+    )
     @console_ns.response(200, "Completion generated successfully")
     @console_ns.response(400, "Invalid request parameters")
     @console_ns.response(404, "App not found")
@@ -89,10 +64,18 @@ class CompletionMessageApi(Resource):
     @account_initialization_required
     @get_app_model(mode=AppMode.COMPLETION)
     def post(self, app_model):
-        args_model = CompletionMessagePayload.model_validate(console_ns.payload)
-        args = args_model.model_dump(exclude_none=True, by_alias=True)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("inputs", type=dict, required=True, location="json")
+            .add_argument("query", type=str, location="json", default="")
+            .add_argument("files", type=list, required=False, location="json")
+            .add_argument("model_config", type=dict, required=True, location="json")
+            .add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
+            .add_argument("retriever_from", type=str, required=False, default="dev", location="json")
+        )
+        args = parser.parse_args()
 
-        streaming = args_model.response_mode != "blocking"
+        streaming = args["response_mode"] != "blocking"
         args["auto_generate_name"] = False
 
         try:
@@ -154,7 +137,21 @@ class ChatMessageApi(Resource):
     @console_ns.doc("create_chat_message")
     @console_ns.doc(description="Generate chat message for debugging")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[ChatMessagePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "ChatMessageRequest",
+            {
+                "inputs": fields.Raw(required=True, description="Input variables"),
+                "query": fields.String(required=True, description="User query"),
+                "files": fields.List(fields.Raw(), description="Uploaded files"),
+                "model_config": fields.Raw(required=True, description="Model configuration"),
+                "conversation_id": fields.String(description="Conversation ID"),
+                "parent_message_id": fields.String(description="Parent message ID"),
+                "response_mode": fields.String(enum=["blocking", "streaming"], description="Response mode"),
+                "retriever_from": fields.String(default="dev", description="Retriever source"),
+            },
+        )
+    )
     @console_ns.response(200, "Chat message generated successfully")
     @console_ns.response(400, "Invalid request parameters")
     @console_ns.response(404, "App or conversation not found")
@@ -164,10 +161,20 @@ class ChatMessageApi(Resource):
     @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT])
     @edit_permission_required
     def post(self, app_model):
-        args_model = ChatMessagePayload.model_validate(console_ns.payload)
-        args = args_model.model_dump(exclude_none=True, by_alias=True)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("inputs", type=dict, required=True, location="json")
+            .add_argument("query", type=str, required=True, location="json")
+            .add_argument("files", type=list, required=False, location="json")
+            .add_argument("model_config", type=dict, required=True, location="json")
+            .add_argument("conversation_id", type=uuid_value, location="json")
+            .add_argument("parent_message_id", type=uuid_value, required=False, location="json")
+            .add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
+            .add_argument("retriever_from", type=str, required=False, default="dev", location="json")
+        )
+        args = parser.parse_args()
 
-        streaming = args_model.response_mode != "blocking"
+        streaming = args["response_mode"] != "blocking"
         args["auto_generate_name"] = False
 
         external_trace_id = get_external_trace_id(request)

api/controllers/console/app/conversation.py

@@ -1,9 +1,7 @@
-from typing import Literal
-
 import sqlalchemy as sa
-from flask import abort, request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask import abort
+from flask_restx import Resource, fields, marshal_with, reqparse
+from flask_restx.inputs import int_range
 from sqlalchemy import func, or_
 from sqlalchemy.orm import joinedload
 from werkzeug.exceptions import NotFound
@@ -16,54 +14,13 @@ from extensions.ext_database import db
 from fields.conversation_fields import MessageTextField
 from fields.raws import FilesContainedField
 from libs.datetime_utils import naive_utc_now, parse_time_range
-from libs.helper import TimestampField
+from libs.helper import DatetimeString, TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models import Conversation, EndUser, Message, MessageAnnotation
 from models.model import AppMode
 from services.conversation_service import ConversationService
 from services.errors.conversation import ConversationNotExistsError
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class BaseConversationQuery(BaseModel):
-    keyword: str | None = Field(default=None, description="Search keyword")
-    start: str | None = Field(default=None, description="Start date (YYYY-MM-DD HH:MM)")
-    end: str | None = Field(default=None, description="End date (YYYY-MM-DD HH:MM)")
-    annotation_status: Literal["annotated", "not_annotated", "all"] = Field(
-        default="all", description="Annotation status filter"
-    )
-    page: int = Field(default=1, ge=1, le=99999, description="Page number")
-    limit: int = Field(default=20, ge=1, le=100, description="Page size (1-100)")
-
-    @field_validator("start", "end", mode="before")
-    @classmethod
-    def blank_to_none(cls, value: str | None) -> str | None:
-        if value == "":
-            return None
-        return value
-
-
-class CompletionConversationQuery(BaseConversationQuery):
-    pass
-
-
-class ChatConversationQuery(BaseConversationQuery):
-    message_count_gte: int | None = Field(default=None, ge=1, description="Minimum message count")
-    sort_by: Literal["created_at", "-created_at", "updated_at", "-updated_at"] = Field(
-        default="-updated_at", description="Sort field and direction"
-    )
-
-
-console_ns.schema_model(
-    CompletionConversationQuery.__name__,
-    CompletionConversationQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-console_ns.schema_model(
-    ChatConversationQuery.__name__,
-    ChatConversationQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
 
@@ -326,7 +283,22 @@ class CompletionConversationApi(Resource):
     @console_ns.doc("list_completion_conversations")
     @console_ns.doc(description="Get completion conversations with pagination and filtering")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[CompletionConversationQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("keyword", type=str, location="args", help="Search keyword")
+        .add_argument("start", type=str, location="args", help="Start date (YYYY-MM-DD HH:MM)")
+        .add_argument("end", type=str, location="args", help="End date (YYYY-MM-DD HH:MM)")
+        .add_argument(
+            "annotation_status",
+            type=str,
+            location="args",
+            choices=["annotated", "not_annotated", "all"],
+            default="all",
+            help="Annotation status filter",
+        )
+        .add_argument("page", type=int, location="args", default=1, help="Page number")
+        .add_argument("limit", type=int, location="args", default=20, help="Page size (1-100)")
+    )
     @console_ns.response(200, "Success", conversation_pagination_model)
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -337,17 +309,32 @@ class CompletionConversationApi(Resource):
     @edit_permission_required
     def get(self, app_model):
         current_user, _ = current_account_with_tenant()
-        args = CompletionConversationQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("keyword", type=str, location="args")
+            .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument(
+                "annotation_status",
+                type=str,
+                choices=["annotated", "not_annotated", "all"],
+                default="all",
+                location="args",
+            )
+            .add_argument("page", type=int_range(1, 99999), default=1, location="args")
+            .add_argument("limit", type=int_range(1, 100), default=20, location="args")
+        )
+        args = parser.parse_args()
 
         query = sa.select(Conversation).where(
             Conversation.app_id == app_model.id, Conversation.mode == "completion", Conversation.is_deleted.is_(False)
         )
 
-        if args.keyword:
+        if args["keyword"]:
             query = query.join(Message, Message.conversation_id == Conversation.id).where(
                 or_(
-                    Message.query.ilike(f"%{args.keyword}%"),
-                    Message.answer.ilike(f"%{args.keyword}%"),
+                    Message.query.ilike(f"%{args['keyword']}%"),
+                    Message.answer.ilike(f"%{args['keyword']}%"),
                 )
             )
 
@@ -355,7 +342,7 @@ class CompletionConversationApi(Resource):
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -367,11 +354,11 @@ class CompletionConversationApi(Resource):
             query = query.where(Conversation.created_at < end_datetime_utc)
 
         # FIXME, the type ignore in this file
-        if args.annotation_status == "annotated":
+        if args["annotation_status"] == "annotated":
             query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
                 MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
             )
-        elif args.annotation_status == "not_annotated":
+        elif args["annotation_status"] == "not_annotated":
             query = (
                 query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
                 .group_by(Conversation.id)
@@ -380,7 +367,7 @@ class CompletionConversationApi(Resource):
 
         query = query.order_by(Conversation.created_at.desc())
 
-        conversations = db.paginate(query, page=args.page, per_page=args.limit, error_out=False)
+        conversations = db.paginate(query, page=args["page"], per_page=args["limit"], error_out=False)
 
         return conversations
 
@@ -432,7 +419,31 @@ class ChatConversationApi(Resource):
     @console_ns.doc("list_chat_conversations")
     @console_ns.doc(description="Get chat conversations with pagination, filtering and summary")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[ChatConversationQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("keyword", type=str, location="args", help="Search keyword")
+        .add_argument("start", type=str, location="args", help="Start date (YYYY-MM-DD HH:MM)")
+        .add_argument("end", type=str, location="args", help="End date (YYYY-MM-DD HH:MM)")
+        .add_argument(
+            "annotation_status",
+            type=str,
+            location="args",
+            choices=["annotated", "not_annotated", "all"],
+            default="all",
+            help="Annotation status filter",
+        )
+        .add_argument("message_count_gte", type=int, location="args", help="Minimum message count")
+        .add_argument("page", type=int, location="args", default=1, help="Page number")
+        .add_argument("limit", type=int, location="args", default=20, help="Page size (1-100)")
+        .add_argument(
+            "sort_by",
+            type=str,
+            location="args",
+            choices=["created_at", "-created_at", "updated_at", "-updated_at"],
+            default="-updated_at",
+            help="Sort field and direction",
+        )
+    )
     @console_ns.response(200, "Success", conversation_with_summary_pagination_model)
     @console_ns.response(403, "Insufficient permissions")
     @setup_required
@@ -443,7 +454,31 @@ class ChatConversationApi(Resource):
     @edit_permission_required
     def get(self, app_model):
         current_user, _ = current_account_with_tenant()
-        args = ChatConversationQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("keyword", type=str, location="args")
+            .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument(
+                "annotation_status",
+                type=str,
+                choices=["annotated", "not_annotated", "all"],
+                default="all",
+                location="args",
+            )
+            .add_argument("message_count_gte", type=int_range(1, 99999), required=False, location="args")
+            .add_argument("page", type=int_range(1, 99999), required=False, default=1, location="args")
+            .add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args")
+            .add_argument(
+                "sort_by",
+                type=str,
+                choices=["created_at", "-created_at", "updated_at", "-updated_at"],
+                required=False,
+                default="-updated_at",
+                location="args",
+            )
+        )
+        args = parser.parse_args()
 
         subquery = (
             db.session.query(
@@ -455,8 +490,8 @@ class ChatConversationApi(Resource):
 
         query = sa.select(Conversation).where(Conversation.app_id == app_model.id, Conversation.is_deleted.is_(False))
 
-        if args.keyword:
-            keyword_filter = f"%{args.keyword}%"
+        if args["keyword"]:
+            keyword_filter = f"%{args['keyword']}%"
             query = (
                 query.join(
                     Message,
@@ -479,12 +514,12 @@ class ChatConversationApi(Resource):
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
         if start_datetime_utc:
-            match args.sort_by:
+            match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at >= start_datetime_utc)
                 case "created_at" | "-created_at" | _:
@@ -492,35 +527,35 @@ class ChatConversationApi(Resource):
 
         if end_datetime_utc:
             end_datetime_utc = end_datetime_utc.replace(second=59)
-            match args.sort_by:
+            match args["sort_by"]:
                 case "updated_at" | "-updated_at":
                     query = query.where(Conversation.updated_at <= end_datetime_utc)
                 case "created_at" | "-created_at" | _:
                     query = query.where(Conversation.created_at <= end_datetime_utc)
 
-        if args.annotation_status == "annotated":
+        if args["annotation_status"] == "annotated":
             query = query.options(joinedload(Conversation.message_annotations)).join(  # type: ignore
                 MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
             )
-        elif args.annotation_status == "not_annotated":
+        elif args["annotation_status"] == "not_annotated":
             query = (
                 query.outerjoin(MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id)
                 .group_by(Conversation.id)
                 .having(func.count(MessageAnnotation.id) == 0)
             )
 
-        if args.message_count_gte and args.message_count_gte >= 1:
+        if args["message_count_gte"] and args["message_count_gte"] >= 1:
             query = (
                 query.options(joinedload(Conversation.messages))  # type: ignore
                 .join(Message, Message.conversation_id == Conversation.id)
                 .group_by(Conversation.id)
-                .having(func.count(Message.id) >= args.message_count_gte)
+                .having(func.count(Message.id) >= args["message_count_gte"])
             )
 
         if app_model.mode == AppMode.ADVANCED_CHAT:
             query = query.where(Conversation.invoke_from != InvokeFrom.DEBUGGER)
 
-        match args.sort_by:
+        match args["sort_by"]:
             case "created_at":
                 query = query.order_by(Conversation.created_at.asc())
             case "-created_at":
@@ -532,7 +567,7 @@ class ChatConversationApi(Resource):
             case _:
                 query = query.order_by(Conversation.created_at.desc())
 
-        conversations = db.paginate(query, page=args.page, per_page=args.limit, error_out=False)
+        conversations = db.paginate(query, page=args["page"], per_page=args["limit"], error_out=False)
 
         return conversations
 

api/controllers/console/app/conversation_variables.py

@@ -1,6 +1,4 @@
-from flask import request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field
+from flask_restx import Resource, fields, marshal_with, reqparse
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -16,18 +14,6 @@ from libs.login import login_required
 from models import ConversationVariable
 from models.model import AppMode
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class ConversationVariablesQuery(BaseModel):
-    conversation_id: str = Field(..., description="Conversation ID to filter variables")
-
-
-console_ns.schema_model(
-    ConversationVariablesQuery.__name__,
-    ConversationVariablesQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register base model first
 conversation_variable_model = console_ns.model("ConversationVariable", conversation_variable_fields)
@@ -47,7 +33,11 @@ class ConversationVariablesApi(Resource):
     @console_ns.doc("get_conversation_variables")
     @console_ns.doc(description="Get conversation variables for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[ConversationVariablesQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser().add_argument(
+            "conversation_id", type=str, location="args", help="Conversation ID to filter variables"
+        )
+    )
     @console_ns.response(200, "Conversation variables retrieved successfully", paginated_conversation_variable_model)
     @setup_required
     @login_required
@@ -55,14 +45,18 @@ class ConversationVariablesApi(Resource):
     @get_app_model(mode=AppMode.ADVANCED_CHAT)
     @marshal_with(paginated_conversation_variable_model)
     def get(self, app_model):
-        args = ConversationVariablesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = reqparse.RequestParser().add_argument("conversation_id", type=str, location="args")
+        args = parser.parse_args()
 
         stmt = (
             select(ConversationVariable)
             .where(ConversationVariable.app_id == app_model.id)
             .order_by(ConversationVariable.created_at)
         )
-        stmt = stmt.where(ConversationVariable.conversation_id == args.conversation_id)
+        if args["conversation_id"]:
+            stmt = stmt.where(ConversationVariable.conversation_id == args["conversation_id"])
+        else:
+            raise ValueError("conversation_id is required")
 
         # NOTE: This is a temporary solution to avoid performance issues.
         page = 1

api/controllers/console/app/generator.py

@@ -1,8 +1,6 @@
 from collections.abc import Sequence
-from typing import Any
 
-from flask_restx import Resource
-from pydantic import BaseModel, Field
+from flask_restx import Resource, fields, reqparse
 
 from controllers.console import console_ns
 from controllers.console.app.error import (
@@ -23,54 +21,21 @@ from libs.login import current_account_with_tenant, login_required
 from models import App
 from services.workflow_service import WorkflowService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class RuleGeneratePayload(BaseModel):
-    instruction: str = Field(..., description="Rule generation instruction")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-    no_variable: bool = Field(default=False, description="Whether to exclude variables")
-
-
-class RuleCodeGeneratePayload(RuleGeneratePayload):
-    code_language: str = Field(default="javascript", description="Programming language for code generation")
-
-
-class RuleStructuredOutputPayload(BaseModel):
-    instruction: str = Field(..., description="Structured output generation instruction")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-
-
-class InstructionGeneratePayload(BaseModel):
-    flow_id: str = Field(..., description="Workflow/Flow ID")
-    node_id: str = Field(default="", description="Node ID for workflow context")
-    current: str = Field(default="", description="Current instruction text")
-    language: str = Field(default="javascript", description="Programming language (javascript/python)")
-    instruction: str = Field(..., description="Instruction for generation")
-    model_config_data: dict[str, Any] = Field(..., alias="model_config", description="Model configuration")
-    ideal_output: str = Field(default="", description="Expected ideal output")
-
-
-class InstructionTemplatePayload(BaseModel):
-    type: str = Field(..., description="Instruction template type")
-
-
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(RuleGeneratePayload)
-reg(RuleCodeGeneratePayload)
-reg(RuleStructuredOutputPayload)
-reg(InstructionGeneratePayload)
-reg(InstructionTemplatePayload)
-
 
 @console_ns.route("/rule-generate")
 class RuleGenerateApi(Resource):
     @console_ns.doc("generate_rule_config")
     @console_ns.doc(description="Generate rule configuration using LLM")
-    @console_ns.expect(console_ns.models[RuleGeneratePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "RuleGenerateRequest",
+            {
+                "instruction": fields.String(required=True, description="Rule generation instruction"),
+                "model_config": fields.Raw(required=True, description="Model configuration"),
+                "no_variable": fields.Boolean(required=True, default=False, description="Whether to exclude variables"),
+            },
+        )
+    )
     @console_ns.response(200, "Rule configuration generated successfully")
     @console_ns.response(400, "Invalid request parameters")
     @console_ns.response(402, "Provider quota exceeded")
@@ -78,15 +43,21 @@ class RuleGenerateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        args = RuleGeneratePayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("instruction", type=str, required=True, nullable=False, location="json")
+            .add_argument("model_config", type=dict, required=True, nullable=False, location="json")
+            .add_argument("no_variable", type=bool, required=True, default=False, location="json")
+        )
+        args = parser.parse_args()
         _, current_tenant_id = current_account_with_tenant()
 
         try:
             rules = LLMGenerator.generate_rule_config(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                no_variable=args.no_variable,
+                instruction=args["instruction"],
+                model_config=args["model_config"],
+                no_variable=args["no_variable"],
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -104,7 +75,19 @@ class RuleGenerateApi(Resource):
 class RuleCodeGenerateApi(Resource):
     @console_ns.doc("generate_rule_code")
     @console_ns.doc(description="Generate code rules using LLM")
-    @console_ns.expect(console_ns.models[RuleCodeGeneratePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "RuleCodeGenerateRequest",
+            {
+                "instruction": fields.String(required=True, description="Code generation instruction"),
+                "model_config": fields.Raw(required=True, description="Model configuration"),
+                "no_variable": fields.Boolean(required=True, default=False, description="Whether to exclude variables"),
+                "code_language": fields.String(
+                    default="javascript", description="Programming language for code generation"
+                ),
+            },
+        )
+    )
     @console_ns.response(200, "Code rules generated successfully")
     @console_ns.response(400, "Invalid request parameters")
     @console_ns.response(402, "Provider quota exceeded")
@@ -112,15 +95,22 @@ class RuleCodeGenerateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        args = RuleCodeGeneratePayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("instruction", type=str, required=True, nullable=False, location="json")
+            .add_argument("model_config", type=dict, required=True, nullable=False, location="json")
+            .add_argument("no_variable", type=bool, required=True, default=False, location="json")
+            .add_argument("code_language", type=str, required=False, default="javascript", location="json")
+        )
+        args = parser.parse_args()
         _, current_tenant_id = current_account_with_tenant()
 
         try:
             code_result = LLMGenerator.generate_code(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
-                code_language=args.code_language,
+                instruction=args["instruction"],
+                model_config=args["model_config"],
+                code_language=args["code_language"],
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -138,7 +128,15 @@ class RuleCodeGenerateApi(Resource):
 class RuleStructuredOutputGenerateApi(Resource):
     @console_ns.doc("generate_structured_output")
     @console_ns.doc(description="Generate structured output rules using LLM")
-    @console_ns.expect(console_ns.models[RuleStructuredOutputPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "StructuredOutputGenerateRequest",
+            {
+                "instruction": fields.String(required=True, description="Structured output generation instruction"),
+                "model_config": fields.Raw(required=True, description="Model configuration"),
+            },
+        )
+    )
     @console_ns.response(200, "Structured output generated successfully")
     @console_ns.response(400, "Invalid request parameters")
     @console_ns.response(402, "Provider quota exceeded")
@@ -146,14 +144,19 @@ class RuleStructuredOutputGenerateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        args = RuleStructuredOutputPayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("instruction", type=str, required=True, nullable=False, location="json")
+            .add_argument("model_config", type=dict, required=True, nullable=False, location="json")
+        )
+        args = parser.parse_args()
         _, current_tenant_id = current_account_with_tenant()
 
         try:
             structured_output = LLMGenerator.generate_structured_output(
                 tenant_id=current_tenant_id,
-                instruction=args.instruction,
-                model_config=args.model_config_data,
+                instruction=args["instruction"],
+                model_config=args["model_config"],
             )
         except ProviderTokenNotInitError as ex:
             raise ProviderNotInitializeError(ex.description)
@@ -171,7 +174,20 @@ class RuleStructuredOutputGenerateApi(Resource):
 class InstructionGenerateApi(Resource):
     @console_ns.doc("generate_instruction")
     @console_ns.doc(description="Generate instruction for workflow nodes or general use")
-    @console_ns.expect(console_ns.models[InstructionGeneratePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "InstructionGenerateRequest",
+            {
+                "flow_id": fields.String(required=True, description="Workflow/Flow ID"),
+                "node_id": fields.String(description="Node ID for workflow context"),
+                "current": fields.String(description="Current instruction text"),
+                "language": fields.String(default="javascript", description="Programming language (javascript/python)"),
+                "instruction": fields.String(required=True, description="Instruction for generation"),
+                "model_config": fields.Raw(required=True, description="Model configuration"),
+                "ideal_output": fields.String(description="Expected ideal output"),
+            },
+        )
+    )
     @console_ns.response(200, "Instruction generated successfully")
     @console_ns.response(400, "Invalid request parameters or flow/workflow not found")
     @console_ns.response(402, "Provider quota exceeded")
@@ -179,69 +195,79 @@ class InstructionGenerateApi(Resource):
     @login_required
     @account_initialization_required
     def post(self):
-        args = InstructionGeneratePayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("flow_id", type=str, required=True, default="", location="json")
+            .add_argument("node_id", type=str, required=False, default="", location="json")
+            .add_argument("current", type=str, required=False, default="", location="json")
+            .add_argument("language", type=str, required=False, default="javascript", location="json")
+            .add_argument("instruction", type=str, required=True, nullable=False, location="json")
+            .add_argument("model_config", type=dict, required=True, nullable=False, location="json")
+            .add_argument("ideal_output", type=str, required=False, default="", location="json")
+        )
+        args = parser.parse_args()
         _, current_tenant_id = current_account_with_tenant()
         providers: list[type[CodeNodeProvider]] = [Python3CodeProvider, JavascriptCodeProvider]
         code_provider: type[CodeNodeProvider] | None = next(
-            (p for p in providers if p.is_accept_language(args.language)), None
+            (p for p in providers if p.is_accept_language(args["language"])), None
         )
         code_template = code_provider.get_default_code() if code_provider else ""
         try:
             # Generate from nothing for a workflow node
-            if (args.current in (code_template, "")) and args.node_id != "":
-                app = db.session.query(App).where(App.id == args.flow_id).first()
+            if (args["current"] == code_template or args["current"] == "") and args["node_id"] != "":
+                app = db.session.query(App).where(App.id == args["flow_id"]).first()
                 if not app:
-                    return {"error": f"app {args.flow_id} not found"}, 400
+                    return {"error": f"app {args['flow_id']} not found"}, 400
                 workflow = WorkflowService().get_draft_workflow(app_model=app)
                 if not workflow:
-                    return {"error": f"workflow {args.flow_id} not found"}, 400
+                    return {"error": f"workflow {args['flow_id']} not found"}, 400
                 nodes: Sequence = workflow.graph_dict["nodes"]
-                node = [node for node in nodes if node["id"] == args.node_id]
+                node = [node for node in nodes if node["id"] == args["node_id"]]
                 if len(node) == 0:
-                    return {"error": f"node {args.node_id} not found"}, 400
+                    return {"error": f"node {args['node_id']} not found"}, 400
                 node_type = node[0]["data"]["type"]
                 match node_type:
                     case "llm":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
+                            instruction=args["instruction"],
+                            model_config=args["model_config"],
                             no_variable=True,
                         )
                     case "agent":
                         return LLMGenerator.generate_rule_config(
                             current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
+                            instruction=args["instruction"],
+                            model_config=args["model_config"],
                             no_variable=True,
                         )
                     case "code":
                         return LLMGenerator.generate_code(
                             tenant_id=current_tenant_id,
-                            instruction=args.instruction,
-                            model_config=args.model_config_data,
-                            code_language=args.language,
+                            instruction=args["instruction"],
+                            model_config=args["model_config"],
+                            code_language=args["language"],
                         )
                     case _:
                         return {"error": f"invalid node type: {node_type}"}
-            if args.node_id == "" and args.current != "":  # For legacy app without a workflow
+            if args["node_id"] == "" and args["current"] != "":  # For legacy app without a workflow
                 return LLMGenerator.instruction_modify_legacy(
                     tenant_id=current_tenant_id,
-                    flow_id=args.flow_id,
-                    current=args.current,
-                    instruction=args.instruction,
-                    model_config=args.model_config_data,
-                    ideal_output=args.ideal_output,
+                    flow_id=args["flow_id"],
+                    current=args["current"],
+                    instruction=args["instruction"],
+                    model_config=args["model_config"],
+                    ideal_output=args["ideal_output"],
                 )
-            if args.node_id != "" and args.current != "":  # For workflow node
+            if args["node_id"] != "" and args["current"] != "":  # For workflow node
                 return LLMGenerator.instruction_modify_workflow(
                     tenant_id=current_tenant_id,
-                    flow_id=args.flow_id,
-                    node_id=args.node_id,
-                    current=args.current,
-                    instruction=args.instruction,
-                    model_config=args.model_config_data,
-                    ideal_output=args.ideal_output,
+                    flow_id=args["flow_id"],
+                    node_id=args["node_id"],
+                    current=args["current"],
+                    instruction=args["instruction"],
+                    model_config=args["model_config"],
+                    ideal_output=args["ideal_output"],
                     workflow_service=WorkflowService(),
                 )
             return {"error": "incompatible parameters"}, 400
@@ -259,15 +285,24 @@ class InstructionGenerateApi(Resource):
 class InstructionGenerationTemplateApi(Resource):
     @console_ns.doc("get_instruction_template")
     @console_ns.doc(description="Get instruction generation template")
-    @console_ns.expect(console_ns.models[InstructionTemplatePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "InstructionTemplateRequest",
+            {
+                "instruction": fields.String(required=True, description="Template instruction"),
+                "ideal_output": fields.String(description="Expected ideal output"),
+            },
+        )
+    )
     @console_ns.response(200, "Template retrieved successfully")
     @console_ns.response(400, "Invalid request parameters")
     @setup_required
     @login_required
     @account_initialization_required
     def post(self):
-        args = InstructionTemplatePayload.model_validate(console_ns.payload)
-        match args.type:
+        parser = reqparse.RequestParser().add_argument("type", type=str, required=True, default=False, location="json")
+        args = parser.parse_args()
+        match args["type"]:
             case "prompt":
                 from core.llm_generator.prompts import INSTRUCTION_GENERATE_TEMPLATE_PROMPT
 
@@ -277,4 +312,4 @@ class InstructionGenerationTemplateApi(Resource):
 
                 return {"data": INSTRUCTION_GENERATE_TEMPLATE_CODE}
             case _:
-                raise ValueError(f"Invalid type: {args.type}")
+                raise ValueError(f"Invalid type: {args['type']}")

api/controllers/console/app/message.py

@@ -1,9 +1,7 @@
 import logging
-from typing import Literal
 
-from flask import request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource, fields, marshal_with, reqparse
+from flask_restx.inputs import int_range
 from sqlalchemy import exists, select
 from werkzeug.exceptions import InternalServerError, NotFound
 
@@ -35,67 +33,6 @@ from services.errors.message import MessageNotExistsError, SuggestedQuestionsAft
 from services.message_service import MessageService
 
 logger = logging.getLogger(__name__)
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class ChatMessagesQuery(BaseModel):
-    conversation_id: str = Field(..., description="Conversation ID")
-    first_id: str | None = Field(default=None, description="First message ID for pagination")
-    limit: int = Field(default=20, ge=1, le=100, description="Number of messages to return (1-100)")
-
-    @field_validator("first_id", mode="before")
-    @classmethod
-    def empty_to_none(cls, value: str | None) -> str | None:
-        if value == "":
-            return None
-        return value
-
-    @field_validator("conversation_id", "first_id")
-    @classmethod
-    def validate_uuid(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return uuid_value(value)
-
-
-class MessageFeedbackPayload(BaseModel):
-    message_id: str = Field(..., description="Message ID")
-    rating: Literal["like", "dislike"] | None = Field(default=None, description="Feedback rating")
-
-    @field_validator("message_id")
-    @classmethod
-    def validate_message_id(cls, value: str) -> str:
-        return uuid_value(value)
-
-
-class FeedbackExportQuery(BaseModel):
-    from_source: Literal["user", "admin"] | None = Field(default=None, description="Filter by feedback source")
-    rating: Literal["like", "dislike"] | None = Field(default=None, description="Filter by rating")
-    has_comment: bool | None = Field(default=None, description="Only include feedback with comments")
-    start_date: str | None = Field(default=None, description="Start date (YYYY-MM-DD)")
-    end_date: str | None = Field(default=None, description="End date (YYYY-MM-DD)")
-    format: Literal["csv", "json"] = Field(default="csv", description="Export format")
-
-    @field_validator("has_comment", mode="before")
-    @classmethod
-    def parse_bool(cls, value: bool | str | None) -> bool | None:
-        if isinstance(value, bool) or value is None:
-            return value
-        lowered = value.lower()
-        if lowered in {"true", "1", "yes", "on"}:
-            return True
-        if lowered in {"false", "0", "no", "off"}:
-            return False
-        raise ValueError("has_comment must be a boolean value")
-
-
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(ChatMessagesQuery)
-reg(MessageFeedbackPayload)
-reg(FeedbackExportQuery)
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -220,7 +157,12 @@ class ChatMessageListApi(Resource):
     @console_ns.doc("list_chat_messages")
     @console_ns.doc(description="Get chat messages for a conversation with pagination")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[ChatMessagesQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("conversation_id", type=str, required=True, location="args", help="Conversation ID")
+        .add_argument("first_id", type=str, location="args", help="First message ID for pagination")
+        .add_argument("limit", type=int, location="args", default=20, help="Number of messages to return (1-100)")
+    )
     @console_ns.response(200, "Success", message_infinite_scroll_pagination_model)
     @console_ns.response(404, "Conversation not found")
     @login_required
@@ -230,21 +172,27 @@ class ChatMessageListApi(Resource):
     @marshal_with(message_infinite_scroll_pagination_model)
     @edit_permission_required
     def get(self, app_model):
-        args = ChatMessagesQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("conversation_id", required=True, type=uuid_value, location="args")
+            .add_argument("first_id", type=uuid_value, location="args")
+            .add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args")
+        )
+        args = parser.parse_args()
 
         conversation = (
             db.session.query(Conversation)
-            .where(Conversation.id == args.conversation_id, Conversation.app_id == app_model.id)
+            .where(Conversation.id == args["conversation_id"], Conversation.app_id == app_model.id)
             .first()
         )
 
         if not conversation:
             raise NotFound("Conversation Not Exists.")
 
-        if args.first_id:
+        if args["first_id"]:
             first_message = (
                 db.session.query(Message)
-                .where(Message.conversation_id == conversation.id, Message.id == args.first_id)
+                .where(Message.conversation_id == conversation.id, Message.id == args["first_id"])
                 .first()
             )
 
@@ -259,7 +207,7 @@ class ChatMessageListApi(Resource):
                     Message.id != first_message.id,
                 )
                 .order_by(Message.created_at.desc())
-                .limit(args.limit)
+                .limit(args["limit"])
                 .all()
             )
         else:
@@ -267,12 +215,12 @@ class ChatMessageListApi(Resource):
                 db.session.query(Message)
                 .where(Message.conversation_id == conversation.id)
                 .order_by(Message.created_at.desc())
-                .limit(args.limit)
+                .limit(args["limit"])
                 .all()
             )
 
         # Initialize has_more based on whether we have a full page
-        if len(history_messages) == args.limit:
+        if len(history_messages) == args["limit"]:
             current_page_first_message = history_messages[-1]
             # Check if there are more messages before the current page
             has_more = db.session.scalar(
@@ -290,7 +238,7 @@ class ChatMessageListApi(Resource):
 
         history_messages = list(reversed(history_messages))
 
-        return InfiniteScrollPagination(data=history_messages, limit=args.limit, has_more=has_more)
+        return InfiniteScrollPagination(data=history_messages, limit=args["limit"], has_more=has_more)
 
 
 @console_ns.route("/apps/<uuid:app_id>/feedbacks")
@@ -298,7 +246,15 @@ class MessageFeedbackApi(Resource):
     @console_ns.doc("create_message_feedback")
     @console_ns.doc(description="Create or update message feedback (like/dislike)")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[MessageFeedbackPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "MessageFeedbackRequest",
+            {
+                "message_id": fields.String(required=True, description="Message ID"),
+                "rating": fields.String(enum=["like", "dislike"], description="Feedback rating"),
+            },
+        )
+    )
     @console_ns.response(200, "Feedback updated successfully")
     @console_ns.response(404, "Message not found")
     @console_ns.response(403, "Insufficient permissions")
@@ -309,9 +265,14 @@ class MessageFeedbackApi(Resource):
     def post(self, app_model):
         current_user, _ = current_account_with_tenant()
 
-        args = MessageFeedbackPayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("message_id", required=True, type=uuid_value, location="json")
+            .add_argument("rating", type=str, choices=["like", "dislike", None], location="json")
+        )
+        args = parser.parse_args()
 
-        message_id = str(args.message_id)
+        message_id = str(args["message_id"])
 
         message = db.session.query(Message).where(Message.id == message_id, Message.app_id == app_model.id).first()
 
@@ -320,21 +281,18 @@ class MessageFeedbackApi(Resource):
 
         feedback = message.admin_feedback
 
-        if not args.rating and feedback:
+        if not args["rating"] and feedback:
             db.session.delete(feedback)
-        elif args.rating and feedback:
-            feedback.rating = args.rating
-        elif not args.rating and not feedback:
+        elif args["rating"] and feedback:
+            feedback.rating = args["rating"]
+        elif not args["rating"] and not feedback:
             raise ValueError("rating cannot be None when feedback not exists")
         else:
-            rating_value = args.rating
-            if rating_value is None:
-                raise ValueError("rating is required to create feedback")
             feedback = MessageFeedback(
                 app_id=app_model.id,
                 conversation_id=message.conversation_id,
                 message_id=message.id,
-                rating=rating_value,
+                rating=args["rating"],
                 from_source="admin",
                 from_account_id=current_user.id,
             )
@@ -411,12 +369,24 @@ class MessageSuggestedQuestionApi(Resource):
         return {"data": questions}
 
 
+# Shared parser for feedback export (used for both documentation and runtime parsing)
+feedback_export_parser = (
+    console_ns.parser()
+    .add_argument("from_source", type=str, choices=["user", "admin"], location="args", help="Filter by feedback source")
+    .add_argument("rating", type=str, choices=["like", "dislike"], location="args", help="Filter by rating")
+    .add_argument("has_comment", type=bool, location="args", help="Only include feedback with comments")
+    .add_argument("start_date", type=str, location="args", help="Start date (YYYY-MM-DD)")
+    .add_argument("end_date", type=str, location="args", help="End date (YYYY-MM-DD)")
+    .add_argument("format", type=str, choices=["csv", "json"], default="csv", location="args", help="Export format")
+)
+
+
 @console_ns.route("/apps/<uuid:app_id>/feedbacks/export")
 class MessageFeedbackExportApi(Resource):
     @console_ns.doc("export_feedbacks")
     @console_ns.doc(description="Export user feedback data for Google Sheets")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[FeedbackExportQuery.__name__])
+    @console_ns.expect(feedback_export_parser)
     @console_ns.response(200, "Feedback data exported successfully")
     @console_ns.response(400, "Invalid parameters")
     @console_ns.response(500, "Internal server error")
@@ -425,7 +395,7 @@ class MessageFeedbackExportApi(Resource):
     @login_required
     @account_initialization_required
     def get(self, app_model):
-        args = FeedbackExportQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = feedback_export_parser.parse_args()
 
         # Import the service function
         from services.feedback_service import FeedbackService
@@ -433,12 +403,12 @@ class MessageFeedbackExportApi(Resource):
         try:
             export_data = FeedbackService.export_feedbacks(
                 app_id=app_model.id,
-                from_source=args.from_source,
-                rating=args.rating,
-                has_comment=args.has_comment,
-                start_date=args.start_date,
-                end_date=args.end_date,
-                format_type=args.format,
+                from_source=args.get("from_source"),
+                rating=args.get("rating"),
+                has_comment=args.get("has_comment"),
+                start_date=args.get("start_date"),
+                end_date=args.get("end_date"),
+                format_type=args.get("format", "csv"),
             )
 
             return export_data

api/controllers/console/app/statistic.py

@@ -1,9 +1,8 @@
 from decimal import Decimal
 
 import sqlalchemy as sa
-from flask import abort, jsonify, request
-from flask_restx import Resource, fields
-from pydantic import BaseModel, Field, field_validator
+from flask import abort, jsonify
+from flask_restx import Resource, fields, reqparse
 
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
@@ -11,37 +10,21 @@ from controllers.console.wraps import account_initialization_required, setup_req
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from libs.datetime_utils import parse_time_range
-from libs.helper import convert_datetime_to_date
+from libs.helper import DatetimeString, convert_datetime_to_date
 from libs.login import current_account_with_tenant, login_required
 from models import AppMode
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class StatisticTimeRangeQuery(BaseModel):
-    start: str | None = Field(default=None, description="Start date (YYYY-MM-DD HH:MM)")
-    end: str | None = Field(default=None, description="End date (YYYY-MM-DD HH:MM)")
-
-    @field_validator("start", "end", mode="before")
-    @classmethod
-    def empty_string_to_none(cls, value: str | None) -> str | None:
-        if value == "":
-            return None
-        return value
-
-
-console_ns.schema_model(
-    StatisticTimeRangeQuery.__name__,
-    StatisticTimeRangeQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
 
 @console_ns.route("/apps/<uuid:app_id>/statistics/daily-messages")
 class DailyMessageStatistic(Resource):
     @console_ns.doc("get_daily_message_statistics")
     @console_ns.doc(description="Get daily message statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("start", type=str, location="args", help="Start date (YYYY-MM-DD HH:MM)")
+        .add_argument("end", type=str, location="args", help="End date (YYYY-MM-DD HH:MM)")
+    )
     @console_ns.response(
         200,
         "Daily message statistics retrieved successfully",
@@ -54,7 +37,12 @@ class DailyMessageStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        )
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -69,7 +57,7 @@ WHERE
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -93,12 +81,19 @@ WHERE
         return jsonify({"data": response_data})
 
 
+parser = (
+    reqparse.RequestParser()
+    .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args", help="Start date (YYYY-MM-DD HH:MM)")
+    .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args", help="End date (YYYY-MM-DD HH:MM)")
+)
+
+
 @console_ns.route("/apps/<uuid:app_id>/statistics/daily-conversations")
 class DailyConversationStatistic(Resource):
     @console_ns.doc("get_daily_conversation_statistics")
     @console_ns.doc(description="Get daily conversation statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200,
         "Daily conversation statistics retrieved successfully",
@@ -111,7 +106,7 @@ class DailyConversationStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -126,7 +121,7 @@ WHERE
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -154,7 +149,7 @@ class DailyTerminalsStatistic(Resource):
     @console_ns.doc("get_daily_terminals_statistics")
     @console_ns.doc(description="Get daily terminal/end-user statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200,
         "Daily terminal statistics retrieved successfully",
@@ -167,7 +162,7 @@ class DailyTerminalsStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -182,7 +177,7 @@ WHERE
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -211,7 +206,7 @@ class DailyTokenCostStatistic(Resource):
     @console_ns.doc("get_daily_token_cost_statistics")
     @console_ns.doc(description="Get daily token cost statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200,
         "Daily token cost statistics retrieved successfully",
@@ -224,7 +219,7 @@ class DailyTokenCostStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -240,7 +235,7 @@ WHERE
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -271,7 +266,7 @@ class AverageSessionInteractionStatistic(Resource):
     @console_ns.doc("get_average_session_interaction_statistics")
     @console_ns.doc(description="Get average session interaction statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200,
         "Average session interaction statistics retrieved successfully",
@@ -284,7 +279,7 @@ class AverageSessionInteractionStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("c.created_at")
         sql_query = f"""SELECT
@@ -307,7 +302,7 @@ FROM
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -347,7 +342,7 @@ class UserSatisfactionRateStatistic(Resource):
     @console_ns.doc("get_user_satisfaction_rate_statistics")
     @console_ns.doc(description="Get user satisfaction rate statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200,
         "User satisfaction rate statistics retrieved successfully",
@@ -360,7 +355,7 @@ class UserSatisfactionRateStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("m.created_at")
         sql_query = f"""SELECT
@@ -379,7 +374,7 @@ WHERE
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -413,7 +408,7 @@ class AverageResponseTimeStatistic(Resource):
     @console_ns.doc("get_average_response_time_statistics")
     @console_ns.doc(description="Get average response time statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200,
         "Average response time statistics retrieved successfully",
@@ -426,7 +421,7 @@ class AverageResponseTimeStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -441,7 +436,7 @@ WHERE
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -470,7 +465,7 @@ class TokensPerSecondStatistic(Resource):
     @console_ns.doc("get_tokens_per_second_statistics")
     @console_ns.doc(description="Get tokens per second statistics for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[StatisticTimeRangeQuery.__name__])
+    @console_ns.expect(parser)
     @console_ns.response(
         200,
         "Tokens per second statistics retrieved successfully",
@@ -482,7 +477,7 @@ class TokensPerSecondStatistic(Resource):
     @account_initialization_required
     def get(self, app_model):
         account, _ = current_account_with_tenant()
-        args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser.parse_args()
 
         converted_created_at = convert_datetime_to_date("created_at")
         sql_query = f"""SELECT
@@ -500,7 +495,7 @@ WHERE
         assert account.timezone is not None
 
         try:
-            start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
+            start_datetime_utc, end_datetime_utc = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 

api/controllers/console/app/workflow.py

@@ -1,11 +1,10 @@
 import json
 import logging
 from collections.abc import Sequence
-from typing import Any
+from typing import cast
 
 from flask import abort, request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource, fields, inputs, marshal_with, reqparse
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
@@ -50,7 +49,6 @@ from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseE
 
 logger = logging.getLogger(__name__)
 LISTENING_RETRY_IN = 2000
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 # Register models for flask_restx to avoid dict type issues in Swagger
 # Register in dependency order: base models first, then dependent models
@@ -109,104 +107,6 @@ if workflow_run_node_execution_model is None:
     workflow_run_node_execution_model = console_ns.model("WorkflowRunNodeExecution", workflow_run_node_execution_fields)
 
 
-class SyncDraftWorkflowPayload(BaseModel):
-    graph: dict[str, Any]
-    features: dict[str, Any]
-    hash: str | None = None
-    environment_variables: list[dict[str, Any]] = Field(default_factory=list)
-    conversation_variables: list[dict[str, Any]] = Field(default_factory=list)
-
-
-class BaseWorkflowRunPayload(BaseModel):
-    files: list[dict[str, Any]] | None = None
-
-
-class AdvancedChatWorkflowRunPayload(BaseWorkflowRunPayload):
-    inputs: dict[str, Any] | None = None
-    query: str = ""
-    conversation_id: str | None = None
-    parent_message_id: str | None = None
-
-    @field_validator("conversation_id", "parent_message_id")
-    @classmethod
-    def validate_uuid(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return uuid_value(value)
-
-
-class IterationNodeRunPayload(BaseModel):
-    inputs: dict[str, Any] | None = None
-
-
-class LoopNodeRunPayload(BaseModel):
-    inputs: dict[str, Any] | None = None
-
-
-class DraftWorkflowRunPayload(BaseWorkflowRunPayload):
-    inputs: dict[str, Any]
-
-
-class DraftWorkflowNodeRunPayload(BaseWorkflowRunPayload):
-    inputs: dict[str, Any]
-    query: str = ""
-
-
-class PublishWorkflowPayload(BaseModel):
-    marked_name: str | None = Field(default=None, max_length=20)
-    marked_comment: str | None = Field(default=None, max_length=100)
-
-
-class DefaultBlockConfigQuery(BaseModel):
-    q: str | None = None
-
-
-class ConvertToWorkflowPayload(BaseModel):
-    name: str | None = None
-    icon_type: str | None = None
-    icon: str | None = None
-    icon_background: str | None = None
-
-
-class WorkflowListQuery(BaseModel):
-    page: int = Field(default=1, ge=1, le=99999)
-    limit: int = Field(default=10, ge=1, le=100)
-    user_id: str | None = None
-    named_only: bool = False
-
-
-class WorkflowUpdatePayload(BaseModel):
-    marked_name: str | None = Field(default=None, max_length=20)
-    marked_comment: str | None = Field(default=None, max_length=100)
-
-
-class DraftWorkflowTriggerRunPayload(BaseModel):
-    node_id: str
-
-
-class DraftWorkflowTriggerRunAllPayload(BaseModel):
-    node_ids: list[str]
-
-
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(SyncDraftWorkflowPayload)
-reg(AdvancedChatWorkflowRunPayload)
-reg(IterationNodeRunPayload)
-reg(LoopNodeRunPayload)
-reg(DraftWorkflowRunPayload)
-reg(DraftWorkflowNodeRunPayload)
-reg(PublishWorkflowPayload)
-reg(DefaultBlockConfigQuery)
-reg(ConvertToWorkflowPayload)
-reg(WorkflowListQuery)
-reg(WorkflowUpdatePayload)
-reg(DraftWorkflowTriggerRunPayload)
-reg(DraftWorkflowTriggerRunAllPayload)
-
-
 # TODO(QuantumGhost): Refactor existing node run API to handle file parameter parsing
 # at the controller level rather than in the workflow logic. This would improve separation
 # of concerns and make the code more maintainable.
@@ -258,7 +158,18 @@ class DraftWorkflowApi(Resource):
     @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
     @console_ns.doc("sync_draft_workflow")
     @console_ns.doc(description="Sync draft workflow configuration")
-    @console_ns.expect(console_ns.models[SyncDraftWorkflowPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "SyncDraftWorkflowRequest",
+            {
+                "graph": fields.Raw(required=True, description="Workflow graph configuration"),
+                "features": fields.Raw(required=True, description="Workflow features configuration"),
+                "hash": fields.String(description="Workflow hash for validation"),
+                "environment_variables": fields.List(fields.Raw, required=True, description="Environment variables"),
+                "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
+            },
+        )
+    )
     @console_ns.response(
         200,
         "Draft workflow synced successfully",
@@ -282,23 +193,36 @@ class DraftWorkflowApi(Resource):
 
         content_type = request.headers.get("Content-Type", "")
 
-        payload_data: dict[str, Any] | None = None
         if "application/json" in content_type:
-            payload_data = request.get_json(silent=True)
-            if not isinstance(payload_data, dict):
-                return {"message": "Invalid JSON data"}, 400
+            parser = (
+                reqparse.RequestParser()
+                .add_argument("graph", type=dict, required=True, nullable=False, location="json")
+                .add_argument("features", type=dict, required=True, nullable=False, location="json")
+                .add_argument("hash", type=str, required=False, location="json")
+                .add_argument("environment_variables", type=list, required=True, location="json")
+                .add_argument("conversation_variables", type=list, required=False, location="json")
+            )
+            args = parser.parse_args()
         elif "text/plain" in content_type:
             try:
-                payload_data = json.loads(request.data.decode("utf-8"))
+                data = json.loads(request.data.decode("utf-8"))
+                if "graph" not in data or "features" not in data:
+                    raise ValueError("graph or features not found in data")
+
+                if not isinstance(data.get("graph"), dict) or not isinstance(data.get("features"), dict):
+                    raise ValueError("graph or features is not a dict")
+
+                args = {
+                    "graph": data.get("graph"),
+                    "features": data.get("features"),
+                    "hash": data.get("hash"),
+                    "environment_variables": data.get("environment_variables"),
+                    "conversation_variables": data.get("conversation_variables"),
+                }
             except json.JSONDecodeError:
                 return {"message": "Invalid JSON data"}, 400
-            if not isinstance(payload_data, dict):
-                return {"message": "Invalid JSON data"}, 400
         else:
             abort(415)
-
-        args_model = SyncDraftWorkflowPayload.model_validate(payload_data)
-        args = args_model.model_dump()
         workflow_service = WorkflowService()
 
         try:
@@ -334,7 +258,17 @@ class AdvancedChatDraftWorkflowRunApi(Resource):
     @console_ns.doc("run_advanced_chat_draft_workflow")
     @console_ns.doc(description="Run draft workflow for advanced chat application")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[AdvancedChatWorkflowRunPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "AdvancedChatWorkflowRunRequest",
+            {
+                "query": fields.String(required=True, description="User query"),
+                "inputs": fields.Raw(description="Input variables"),
+                "files": fields.List(fields.Raw, description="File uploads"),
+                "conversation_id": fields.String(description="Conversation ID"),
+            },
+        )
+    )
     @console_ns.response(200, "Workflow run started successfully")
     @console_ns.response(400, "Invalid request parameters")
     @console_ns.response(403, "Permission denied")
@@ -349,8 +283,16 @@ class AdvancedChatDraftWorkflowRunApi(Resource):
         """
         current_user, _ = current_account_with_tenant()
 
-        args_model = AdvancedChatWorkflowRunPayload.model_validate(console_ns.payload or {})
-        args = args_model.model_dump(exclude_none=True)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("inputs", type=dict, location="json")
+            .add_argument("query", type=str, required=True, location="json", default="")
+            .add_argument("files", type=list, location="json")
+            .add_argument("conversation_id", type=uuid_value, location="json")
+            .add_argument("parent_message_id", type=uuid_value, required=False, location="json")
+        )
+
+        args = parser.parse_args()
 
         external_trace_id = get_external_trace_id(request)
         if external_trace_id:
@@ -380,7 +322,15 @@ class AdvancedChatDraftRunIterationNodeApi(Resource):
     @console_ns.doc("run_advanced_chat_draft_iteration_node")
     @console_ns.doc(description="Run draft workflow iteration node for advanced chat")
     @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @console_ns.expect(console_ns.models[IterationNodeRunPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "IterationNodeRunRequest",
+            {
+                "task_id": fields.String(required=True, description="Task ID"),
+                "inputs": fields.Raw(description="Input variables"),
+            },
+        )
+    )
     @console_ns.response(200, "Iteration node run started successfully")
     @console_ns.response(403, "Permission denied")
     @console_ns.response(404, "Node not found")
@@ -394,7 +344,8 @@ class AdvancedChatDraftRunIterationNodeApi(Resource):
         Run draft workflow iteration node
         """
         current_user, _ = current_account_with_tenant()
-        args = IterationNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+        parser = reqparse.RequestParser().add_argument("inputs", type=dict, location="json")
+        args = parser.parse_args()
 
         try:
             response = AppGenerateService.generate_single_iteration(
@@ -418,7 +369,15 @@ class WorkflowDraftRunIterationNodeApi(Resource):
     @console_ns.doc("run_workflow_draft_iteration_node")
     @console_ns.doc(description="Run draft workflow iteration node")
     @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @console_ns.expect(console_ns.models[IterationNodeRunPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "WorkflowIterationNodeRunRequest",
+            {
+                "task_id": fields.String(required=True, description="Task ID"),
+                "inputs": fields.Raw(description="Input variables"),
+            },
+        )
+    )
     @console_ns.response(200, "Workflow iteration node run started successfully")
     @console_ns.response(403, "Permission denied")
     @console_ns.response(404, "Node not found")
@@ -432,7 +391,8 @@ class WorkflowDraftRunIterationNodeApi(Resource):
         Run draft workflow iteration node
         """
         current_user, _ = current_account_with_tenant()
-        args = IterationNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+        parser = reqparse.RequestParser().add_argument("inputs", type=dict, location="json")
+        args = parser.parse_args()
 
         try:
             response = AppGenerateService.generate_single_iteration(
@@ -456,7 +416,15 @@ class AdvancedChatDraftRunLoopNodeApi(Resource):
     @console_ns.doc("run_advanced_chat_draft_loop_node")
     @console_ns.doc(description="Run draft workflow loop node for advanced chat")
     @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @console_ns.expect(console_ns.models[LoopNodeRunPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "LoopNodeRunRequest",
+            {
+                "task_id": fields.String(required=True, description="Task ID"),
+                "inputs": fields.Raw(description="Input variables"),
+            },
+        )
+    )
     @console_ns.response(200, "Loop node run started successfully")
     @console_ns.response(403, "Permission denied")
     @console_ns.response(404, "Node not found")
@@ -470,7 +438,8 @@ class AdvancedChatDraftRunLoopNodeApi(Resource):
         Run draft workflow loop node
         """
         current_user, _ = current_account_with_tenant()
-        args = LoopNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+        parser = reqparse.RequestParser().add_argument("inputs", type=dict, location="json")
+        args = parser.parse_args()
 
         try:
             response = AppGenerateService.generate_single_loop(
@@ -494,7 +463,15 @@ class WorkflowDraftRunLoopNodeApi(Resource):
     @console_ns.doc("run_workflow_draft_loop_node")
     @console_ns.doc(description="Run draft workflow loop node")
     @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @console_ns.expect(console_ns.models[LoopNodeRunPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "WorkflowLoopNodeRunRequest",
+            {
+                "task_id": fields.String(required=True, description="Task ID"),
+                "inputs": fields.Raw(description="Input variables"),
+            },
+        )
+    )
     @console_ns.response(200, "Workflow loop node run started successfully")
     @console_ns.response(403, "Permission denied")
     @console_ns.response(404, "Node not found")
@@ -508,7 +485,8 @@ class WorkflowDraftRunLoopNodeApi(Resource):
         Run draft workflow loop node
         """
         current_user, _ = current_account_with_tenant()
-        args = LoopNodeRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+        parser = reqparse.RequestParser().add_argument("inputs", type=dict, location="json")
+        args = parser.parse_args()
 
         try:
             response = AppGenerateService.generate_single_loop(
@@ -532,7 +510,15 @@ class DraftWorkflowRunApi(Resource):
     @console_ns.doc("run_draft_workflow")
     @console_ns.doc(description="Run draft workflow")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[DraftWorkflowRunPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "DraftWorkflowRunRequest",
+            {
+                "inputs": fields.Raw(required=True, description="Input variables"),
+                "files": fields.List(fields.Raw, description="File uploads"),
+            },
+        )
+    )
     @console_ns.response(200, "Draft workflow run started successfully")
     @console_ns.response(403, "Permission denied")
     @setup_required
@@ -545,7 +531,12 @@ class DraftWorkflowRunApi(Resource):
         Run draft workflow
         """
         current_user, _ = current_account_with_tenant()
-        args = DraftWorkflowRunPayload.model_validate(console_ns.payload or {}).model_dump(exclude_none=True)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+            .add_argument("files", type=list, required=False, location="json")
+        )
+        args = parser.parse_args()
 
         external_trace_id = get_external_trace_id(request)
         if external_trace_id:
@@ -597,7 +588,14 @@ class DraftWorkflowNodeRunApi(Resource):
     @console_ns.doc("run_draft_workflow_node")
     @console_ns.doc(description="Run draft workflow node")
     @console_ns.doc(params={"app_id": "Application ID", "node_id": "Node ID"})
-    @console_ns.expect(console_ns.models[DraftWorkflowNodeRunPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "DraftWorkflowNodeRunRequest",
+            {
+                "inputs": fields.Raw(description="Input variables"),
+            },
+        )
+    )
     @console_ns.response(200, "Node run started successfully", workflow_run_node_execution_model)
     @console_ns.response(403, "Permission denied")
     @console_ns.response(404, "Node not found")
@@ -612,10 +610,15 @@ class DraftWorkflowNodeRunApi(Resource):
         Run draft workflow node
         """
         current_user, _ = current_account_with_tenant()
-        args_model = DraftWorkflowNodeRunPayload.model_validate(console_ns.payload or {})
-        args = args_model.model_dump(exclude_none=True)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+            .add_argument("query", type=str, required=False, location="json", default="")
+            .add_argument("files", type=list, location="json", default=[])
+        )
+        args = parser.parse_args()
 
-        user_inputs = args_model.inputs
+        user_inputs = args.get("inputs")
         if user_inputs is None:
             raise ValueError("missing inputs")
 
@@ -640,6 +643,13 @@ class DraftWorkflowNodeRunApi(Resource):
         return workflow_node_execution
 
 
+parser_publish = (
+    reqparse.RequestParser()
+    .add_argument("marked_name", type=str, required=False, default="", location="json")
+    .add_argument("marked_comment", type=str, required=False, default="", location="json")
+)
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/publish")
 class PublishedWorkflowApi(Resource):
     @console_ns.doc("get_published_workflow")
@@ -664,7 +674,7 @@ class PublishedWorkflowApi(Resource):
         # return workflow, if not found, return None
         return workflow
 
-    @console_ns.expect(console_ns.models[PublishWorkflowPayload.__name__])
+    @console_ns.expect(parser_publish)
     @setup_required
     @login_required
     @account_initialization_required
@@ -676,7 +686,13 @@ class PublishedWorkflowApi(Resource):
         """
         current_user, _ = current_account_with_tenant()
 
-        args = PublishWorkflowPayload.model_validate(console_ns.payload or {})
+        args = parser_publish.parse_args()
+
+        # Validate name and comment length
+        if args.marked_name and len(args.marked_name) > 20:
+            raise ValueError("Marked name cannot exceed 20 characters")
+        if args.marked_comment and len(args.marked_comment) > 100:
+            raise ValueError("Marked comment cannot exceed 100 characters")
 
         workflow_service = WorkflowService()
         with Session(db.engine) as session:
@@ -725,6 +741,9 @@ class DefaultBlockConfigsApi(Resource):
         return workflow_service.get_default_block_configs()
 
 
+parser_block = reqparse.RequestParser().add_argument("q", type=str, location="args")
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>")
 class DefaultBlockConfigApi(Resource):
     @console_ns.doc("get_default_block_config")
@@ -732,7 +751,7 @@ class DefaultBlockConfigApi(Resource):
     @console_ns.doc(params={"app_id": "Application ID", "block_type": "Block type"})
     @console_ns.response(200, "Default block configuration retrieved successfully")
     @console_ns.response(404, "Block type not found")
-    @console_ns.expect(console_ns.models[DefaultBlockConfigQuery.__name__])
+    @console_ns.expect(parser_block)
     @setup_required
     @login_required
     @account_initialization_required
@@ -742,12 +761,14 @@ class DefaultBlockConfigApi(Resource):
         """
         Get default block config
         """
-        args = DefaultBlockConfigQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        args = parser_block.parse_args()
+
+        q = args.get("q")
 
         filters = None
-        if args.q:
+        if q:
             try:
-                filters = json.loads(args.q)
+                filters = json.loads(args.get("q", ""))
             except json.JSONDecodeError:
                 raise ValueError("Invalid filters")
 
@@ -756,9 +777,18 @@ class DefaultBlockConfigApi(Resource):
         return workflow_service.get_default_block_config(node_type=block_type, filters=filters)
 
 
+parser_convert = (
+    reqparse.RequestParser()
+    .add_argument("name", type=str, required=False, nullable=True, location="json")
+    .add_argument("icon_type", type=str, required=False, nullable=True, location="json")
+    .add_argument("icon", type=str, required=False, nullable=True, location="json")
+    .add_argument("icon_background", type=str, required=False, nullable=True, location="json")
+)
+
+
 @console_ns.route("/apps/<uuid:app_id>/convert-to-workflow")
 class ConvertToWorkflowApi(Resource):
-    @console_ns.expect(console_ns.models[ConvertToWorkflowPayload.__name__])
+    @console_ns.expect(parser_convert)
     @console_ns.doc("convert_to_workflow")
     @console_ns.doc(description="Convert application to workflow mode")
     @console_ns.doc(params={"app_id": "Application ID"})
@@ -778,8 +808,10 @@ class ConvertToWorkflowApi(Resource):
         """
         current_user, _ = current_account_with_tenant()
 
-        payload = console_ns.payload or {}
-        args = ConvertToWorkflowPayload.model_validate(payload).model_dump(exclude_none=True)
+        if request.data:
+            args = parser_convert.parse_args()
+        else:
+            args = {}
 
         # convert to workflow mode
         workflow_service = WorkflowService()
@@ -791,9 +823,18 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
+parser_workflows = (
+    reqparse.RequestParser()
+    .add_argument("page", type=inputs.int_range(1, 99999), required=False, default=1, location="args")
+    .add_argument("limit", type=inputs.int_range(1, 100), required=False, default=10, location="args")
+    .add_argument("user_id", type=str, required=False, location="args")
+    .add_argument("named_only", type=inputs.boolean, required=False, default=False, location="args")
+)
+
+
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
-    @console_ns.expect(console_ns.models[WorkflowListQuery.__name__])
+    @console_ns.expect(parser_workflows)
     @console_ns.doc("get_all_published_workflows")
     @console_ns.doc(description="Get all published workflows for an application")
     @console_ns.doc(params={"app_id": "Application ID"})
@@ -810,15 +851,16 @@ class PublishedAllWorkflowApi(Resource):
         """
         current_user, _ = current_account_with_tenant()
 
-        args = WorkflowListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        page = args.page
-        limit = args.limit
-        user_id = args.user_id
-        named_only = args.named_only
+        args = parser_workflows.parse_args()
+        page = args["page"]
+        limit = args["limit"]
+        user_id = args.get("user_id")
+        named_only = args.get("named_only", False)
 
         if user_id:
             if user_id != current_user.id:
                 raise Forbidden()
+            user_id = cast(str, user_id)
 
         workflow_service = WorkflowService()
         with Session(db.engine) as session:
@@ -844,7 +886,15 @@ class WorkflowByIdApi(Resource):
     @console_ns.doc("update_workflow_by_id")
     @console_ns.doc(description="Update workflow by ID")
     @console_ns.doc(params={"app_id": "Application ID", "workflow_id": "Workflow ID"})
-    @console_ns.expect(console_ns.models[WorkflowUpdatePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "UpdateWorkflowRequest",
+            {
+                "environment_variables": fields.List(fields.Raw, description="Environment variables"),
+                "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
+            },
+        )
+    )
     @console_ns.response(200, "Workflow updated successfully", workflow_model)
     @console_ns.response(404, "Workflow not found")
     @console_ns.response(403, "Permission denied")
@@ -859,14 +909,25 @@ class WorkflowByIdApi(Resource):
         Update workflow attributes
         """
         current_user, _ = current_account_with_tenant()
-        args = WorkflowUpdatePayload.model_validate(console_ns.payload or {})
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("marked_name", type=str, required=False, location="json")
+            .add_argument("marked_comment", type=str, required=False, location="json")
+        )
+        args = parser.parse_args()
+
+        # Validate name and comment length
+        if args.marked_name and len(args.marked_name) > 20:
+            raise ValueError("Marked name cannot exceed 20 characters")
+        if args.marked_comment and len(args.marked_comment) > 100:
+            raise ValueError("Marked comment cannot exceed 100 characters")
 
         # Prepare update data
         update_data = {}
-        if args.marked_name is not None:
-            update_data["marked_name"] = args.marked_name
-        if args.marked_comment is not None:
-            update_data["marked_comment"] = args.marked_comment
+        if args.get("marked_name") is not None:
+            update_data["marked_name"] = args["marked_name"]
+        if args.get("marked_comment") is not None:
+            update_data["marked_comment"] = args["marked_comment"]
 
         if not update_data:
             return {"message": "No valid fields to update"}, 400
@@ -979,8 +1040,11 @@ class DraftWorkflowTriggerRunApi(Resource):
         Poll for trigger events and execute full workflow when event arrives
         """
         current_user, _ = current_account_with_tenant()
-        args = DraftWorkflowTriggerRunPayload.model_validate(console_ns.payload or {})
-        node_id = args.node_id
+        parser = reqparse.RequestParser().add_argument(
+            "node_id", type=str, required=True, location="json", nullable=False
+        )
+        args = parser.parse_args()
+        node_id = args["node_id"]
         workflow_service = WorkflowService()
         draft_workflow = workflow_service.get_draft_workflow(app_model)
         if not draft_workflow:
@@ -1108,7 +1172,14 @@ class DraftWorkflowTriggerRunAllApi(Resource):
     @console_ns.doc("draft_workflow_trigger_run_all")
     @console_ns.doc(description="Full workflow debug when the start node is a trigger")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[DraftWorkflowTriggerRunAllPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "DraftWorkflowTriggerRunAllRequest",
+            {
+                "node_ids": fields.List(fields.String, required=True, description="Node IDs"),
+            },
+        )
+    )
     @console_ns.response(200, "Workflow executed successfully")
     @console_ns.response(403, "Permission denied")
     @console_ns.response(500, "Internal server error")
@@ -1123,8 +1194,11 @@ class DraftWorkflowTriggerRunAllApi(Resource):
         """
         current_user, _ = current_account_with_tenant()
 
-        args = DraftWorkflowTriggerRunAllPayload.model_validate(console_ns.payload or {})
-        node_ids = args.node_ids
+        parser = reqparse.RequestParser().add_argument(
+            "node_ids", type=list, required=True, location="json", nullable=False
+        )
+        args = parser.parse_args()
+        node_ids = args["node_ids"]
         workflow_service = WorkflowService()
         draft_workflow = workflow_service.get_draft_workflow(app_model)
         if not draft_workflow:

api/controllers/console/app/workflow_app_log.py

@@ -1,9 +1,6 @@
-from datetime import datetime
-
 from dateutil.parser import isoparse
-from flask import request
-from flask_restx import Resource, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource, marshal_with, reqparse
+from flask_restx.inputs import int_range
 from sqlalchemy.orm import Session
 
 from controllers.console import console_ns
@@ -17,48 +14,6 @@ from models import App
 from models.model import AppMode
 from services.workflow_app_service import WorkflowAppService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class WorkflowAppLogQuery(BaseModel):
-    keyword: str | None = Field(default=None, description="Search keyword for filtering logs")
-    status: WorkflowExecutionStatus | None = Field(
-        default=None, description="Execution status filter (succeeded, failed, stopped, partial-succeeded)"
-    )
-    created_at__before: datetime | None = Field(default=None, description="Filter logs created before this timestamp")
-    created_at__after: datetime | None = Field(default=None, description="Filter logs created after this timestamp")
-    created_by_end_user_session_id: str | None = Field(default=None, description="Filter by end user session ID")
-    created_by_account: str | None = Field(default=None, description="Filter by account")
-    detail: bool = Field(default=False, description="Whether to return detailed logs")
-    page: int = Field(default=1, ge=1, le=99999, description="Page number (1-99999)")
-    limit: int = Field(default=20, ge=1, le=100, description="Number of items per page (1-100)")
-
-    @field_validator("created_at__before", "created_at__after", mode="before")
-    @classmethod
-    def parse_datetime(cls, value: str | None) -> datetime | None:
-        if value in (None, ""):
-            return None
-        return isoparse(value)  # type: ignore
-
-    @field_validator("detail", mode="before")
-    @classmethod
-    def parse_bool(cls, value: bool | str | None) -> bool:
-        if isinstance(value, bool):
-            return value
-        if value is None:
-            return False
-        lowered = value.lower()
-        if lowered in {"1", "true", "yes", "on"}:
-            return True
-        if lowered in {"0", "false", "no", "off"}:
-            return False
-        raise ValueError("Invalid boolean value for detail")
-
-
-console_ns.schema_model(
-    WorkflowAppLogQuery.__name__, WorkflowAppLogQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-
 # Register model for flask_restx to avoid dict type issues in Swagger
 workflow_app_log_pagination_model = build_workflow_app_log_pagination_model(console_ns)
 
@@ -68,7 +23,19 @@ class WorkflowAppLogApi(Resource):
     @console_ns.doc("get_workflow_app_logs")
     @console_ns.doc(description="Get workflow application execution logs")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[WorkflowAppLogQuery.__name__])
+    @console_ns.doc(
+        params={
+            "keyword": "Search keyword for filtering logs",
+            "status": "Filter by execution status (succeeded, failed, stopped, partial-succeeded)",
+            "created_at__before": "Filter logs created before this timestamp",
+            "created_at__after": "Filter logs created after this timestamp",
+            "created_by_end_user_session_id": "Filter by end user session ID",
+            "created_by_account": "Filter by account",
+            "detail": "Whether to return detailed logs",
+            "page": "Page number (1-99999)",
+            "limit": "Number of items per page (1-100)",
+        }
+    )
     @console_ns.response(200, "Workflow app logs retrieved successfully", workflow_app_log_pagination_model)
     @setup_required
     @login_required
@@ -79,7 +46,44 @@ class WorkflowAppLogApi(Resource):
         """
         Get workflow app logs
         """
-        args = WorkflowAppLogQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("keyword", type=str, location="args")
+            .add_argument(
+                "status", type=str, choices=["succeeded", "failed", "stopped", "partial-succeeded"], location="args"
+            )
+            .add_argument(
+                "created_at__before", type=str, location="args", help="Filter logs created before this timestamp"
+            )
+            .add_argument(
+                "created_at__after", type=str, location="args", help="Filter logs created after this timestamp"
+            )
+            .add_argument(
+                "created_by_end_user_session_id",
+                type=str,
+                location="args",
+                required=False,
+                default=None,
+            )
+            .add_argument(
+                "created_by_account",
+                type=str,
+                location="args",
+                required=False,
+                default=None,
+            )
+            .add_argument("detail", type=bool, location="args", required=False, default=False)
+            .add_argument("page", type=int_range(1, 99999), default=1, location="args")
+            .add_argument("limit", type=int_range(1, 100), default=20, location="args")
+        )
+        args = parser.parse_args()
+
+        args.status = WorkflowExecutionStatus(args.status) if args.status else None
+        if args.created_at__before:
+            args.created_at__before = isoparse(args.created_at__before)
+
+        if args.created_at__after:
+            args.created_at__after = isoparse(args.created_at__after)
 
         # get paginate workflow app logs
         workflow_app_service = WorkflowAppService()

api/controllers/console/app/workflow_run.py

@@ -1,8 +1,7 @@
-from typing import Literal, cast
+from typing import cast
 
-from flask import request
-from flask_restx import Resource, fields, marshal_with
-from pydantic import BaseModel, Field, field_validator
+from flask_restx import Resource, fields, marshal_with, reqparse
+from flask_restx.inputs import int_range
 
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
@@ -93,51 +92,70 @@ workflow_run_node_execution_list_model = console_ns.model(
     "WorkflowRunNodeExecutionList", workflow_run_node_execution_list_fields_copy
 )
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
+def _parse_workflow_run_list_args():
+    """
+    Parse common arguments for workflow run list endpoints.
 
-class WorkflowRunListQuery(BaseModel):
-    last_id: str | None = Field(default=None, description="Last run ID for pagination")
-    limit: int = Field(default=20, ge=1, le=100, description="Number of items per page (1-100)")
-    status: Literal["running", "succeeded", "failed", "stopped", "partial-succeeded"] | None = Field(
-        default=None, description="Workflow run status filter"
+    Returns:
+        Parsed arguments containing last_id, limit, status, and triggered_from filters
+    """
+    parser = (
+        reqparse.RequestParser()
+        .add_argument("last_id", type=uuid_value, location="args")
+        .add_argument("limit", type=int_range(1, 100), required=False, default=20, location="args")
+        .add_argument(
+            "status",
+            type=str,
+            choices=WORKFLOW_RUN_STATUS_CHOICES,
+            location="args",
+            required=False,
+        )
+        .add_argument(
+            "triggered_from",
+            type=str,
+            choices=["debugging", "app-run"],
+            location="args",
+            required=False,
+            help="Filter by trigger source: debugging or app-run",
+        )
     )
-    triggered_from: Literal["debugging", "app-run"] | None = Field(
-        default=None, description="Filter by trigger source: debugging or app-run"
+    return parser.parse_args()
+
+
+def _parse_workflow_run_count_args():
+    """
+    Parse common arguments for workflow run count endpoints.
+
+    Returns:
+        Parsed arguments containing status, time_range, and triggered_from filters
+    """
+    parser = (
+        reqparse.RequestParser()
+        .add_argument(
+            "status",
+            type=str,
+            choices=WORKFLOW_RUN_STATUS_CHOICES,
+            location="args",
+            required=False,
+        )
+        .add_argument(
+            "time_range",
+            type=time_duration,
+            location="args",
+            required=False,
+            help="Time range filter (e.g., 7d, 4h, 30m, 30s)",
+        )
+        .add_argument(
+            "triggered_from",
+            type=str,
+            choices=["debugging", "app-run"],
+            location="args",
+            required=False,
+            help="Filter by trigger source: debugging or app-run",
+        )
     )
-
-    @field_validator("last_id")
-    @classmethod
-    def validate_last_id(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return uuid_value(value)
-
-
-class WorkflowRunCountQuery(BaseModel):
-    status: Literal["running", "succeeded", "failed", "stopped", "partial-succeeded"] | None = Field(
-        default=None, description="Workflow run status filter"
-    )
-    time_range: str | None = Field(default=None, description="Time range filter (e.g., 7d, 4h, 30m, 30s)")
-    triggered_from: Literal["debugging", "app-run"] | None = Field(
-        default=None, description="Filter by trigger source: debugging or app-run"
-    )
-
-    @field_validator("time_range")
-    @classmethod
-    def validate_time_range(cls, value: str | None) -> str | None:
-        if value is None:
-            return value
-        return time_duration(value)
-
-
-console_ns.schema_model(
-    WorkflowRunListQuery.__name__, WorkflowRunListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
-)
-console_ns.schema_model(
-    WorkflowRunCountQuery.__name__,
-    WorkflowRunCountQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
+    return parser.parse_args()
 
 
 @console_ns.route("/apps/<uuid:app_id>/advanced-chat/workflow-runs")
@@ -152,7 +170,6 @@ class AdvancedChatAppWorkflowRunListApi(Resource):
     @console_ns.doc(
         params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
     )
-    @console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
     @console_ns.response(200, "Workflow runs retrieved successfully", advanced_chat_workflow_run_pagination_model)
     @setup_required
     @login_required
@@ -163,13 +180,12 @@ class AdvancedChatAppWorkflowRunListApi(Resource):
         """
         Get advanced chat app workflow run list
         """
-        args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        args = args_model.model_dump(exclude_none=True)
+        args = _parse_workflow_run_list_args()
 
         # Default to DEBUGGING if not specified
         triggered_from = (
-            WorkflowRunTriggeredFrom(args_model.triggered_from)
-            if args_model.triggered_from
+            WorkflowRunTriggeredFrom(args.get("triggered_from"))
+            if args.get("triggered_from")
             else WorkflowRunTriggeredFrom.DEBUGGING
         )
 
@@ -201,7 +217,6 @@ class AdvancedChatAppWorkflowRunCountApi(Resource):
         params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
     )
     @console_ns.response(200, "Workflow runs count retrieved successfully", workflow_run_count_model)
-    @console_ns.expect(console_ns.models[WorkflowRunCountQuery.__name__])
     @setup_required
     @login_required
     @account_initialization_required
@@ -211,13 +226,12 @@ class AdvancedChatAppWorkflowRunCountApi(Resource):
         """
         Get advanced chat workflow runs count statistics
         """
-        args_model = WorkflowRunCountQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        args = args_model.model_dump(exclude_none=True)
+        args = _parse_workflow_run_count_args()
 
         # Default to DEBUGGING if not specified
         triggered_from = (
-            WorkflowRunTriggeredFrom(args_model.triggered_from)
-            if args_model.triggered_from
+            WorkflowRunTriggeredFrom(args.get("triggered_from"))
+            if args.get("triggered_from")
             else WorkflowRunTriggeredFrom.DEBUGGING
         )
 
@@ -245,7 +259,6 @@ class WorkflowRunListApi(Resource):
         params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
     )
     @console_ns.response(200, "Workflow runs retrieved successfully", workflow_run_pagination_model)
-    @console_ns.expect(console_ns.models[WorkflowRunListQuery.__name__])
     @setup_required
     @login_required
     @account_initialization_required
@@ -255,13 +268,12 @@ class WorkflowRunListApi(Resource):
         """
         Get workflow run list
         """
-        args_model = WorkflowRunListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        args = args_model.model_dump(exclude_none=True)
+        args = _parse_workflow_run_list_args()
 
         # Default to DEBUGGING for workflow if not specified (backward compatibility)
         triggered_from = (
-            WorkflowRunTriggeredFrom(args_model.triggered_from)
-            if args_model.triggered_from
+            WorkflowRunTriggeredFrom(args.get("triggered_from"))
+            if args.get("triggered_from")
             else WorkflowRunTriggeredFrom.DEBUGGING
         )
 
@@ -293,7 +305,6 @@ class WorkflowRunCountApi(Resource):
         params={"triggered_from": "Filter by trigger source (optional): debugging or app-run. Default: debugging"}
     )
     @console_ns.response(200, "Workflow runs count retrieved successfully", workflow_run_count_model)
-    @console_ns.expect(console_ns.models[WorkflowRunCountQuery.__name__])
     @setup_required
     @login_required
     @account_initialization_required
@@ -303,13 +314,12 @@ class WorkflowRunCountApi(Resource):
         """
         Get workflow runs count statistics
         """
-        args_model = WorkflowRunCountQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
-        args = args_model.model_dump(exclude_none=True)
+        args = _parse_workflow_run_count_args()
 
         # Default to DEBUGGING for workflow if not specified (backward compatibility)
         triggered_from = (
-            WorkflowRunTriggeredFrom(args_model.triggered_from)
-            if args_model.triggered_from
+            WorkflowRunTriggeredFrom(args.get("triggered_from"))
+            if args.get("triggered_from")
             else WorkflowRunTriggeredFrom.DEBUGGING
         )
 

api/controllers/console/app/workflow_statistic.py

@@ -1,6 +1,5 @@
-from flask import abort, jsonify, request
-from flask_restx import Resource
-from pydantic import BaseModel, Field, field_validator
+from flask import abort, jsonify
+from flask_restx import Resource, reqparse
 from sqlalchemy.orm import sessionmaker
 
 from controllers.console import console_ns
@@ -8,31 +7,12 @@ from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from extensions.ext_database import db
 from libs.datetime_utils import parse_time_range
+from libs.helper import DatetimeString
 from libs.login import current_account_with_tenant, login_required
 from models.enums import WorkflowRunTriggeredFrom
 from models.model import AppMode
 from repositories.factory import DifyAPIRepositoryFactory
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class WorkflowStatisticQuery(BaseModel):
-    start: str | None = Field(default=None, description="Start date and time (YYYY-MM-DD HH:MM)")
-    end: str | None = Field(default=None, description="End date and time (YYYY-MM-DD HH:MM)")
-
-    @field_validator("start", "end", mode="before")
-    @classmethod
-    def blank_to_none(cls, value: str | None) -> str | None:
-        if value == "":
-            return None
-        return value
-
-
-console_ns.schema_model(
-    WorkflowStatisticQuery.__name__,
-    WorkflowStatisticQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
-)
-
 
 @console_ns.route("/apps/<uuid:app_id>/workflow/statistics/daily-conversations")
 class WorkflowDailyRunsStatistic(Resource):
@@ -44,7 +24,9 @@ class WorkflowDailyRunsStatistic(Resource):
     @console_ns.doc("get_workflow_daily_runs_statistic")
     @console_ns.doc(description="Get workflow daily runs statistics")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[WorkflowStatisticQuery.__name__])
+    @console_ns.doc(
+        params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"}
+    )
     @console_ns.response(200, "Daily runs statistics retrieved successfully")
     @get_app_model
     @setup_required
@@ -53,12 +35,17 @@ class WorkflowDailyRunsStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        )
+        args = parser.parse_args()
 
         assert account.timezone is not None
 
         try:
-            start_date, end_date = parse_time_range(args.start, args.end, account.timezone)
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -84,7 +71,9 @@ class WorkflowDailyTerminalsStatistic(Resource):
     @console_ns.doc("get_workflow_daily_terminals_statistic")
     @console_ns.doc(description="Get workflow daily terminals statistics")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[WorkflowStatisticQuery.__name__])
+    @console_ns.doc(
+        params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"}
+    )
     @console_ns.response(200, "Daily terminals statistics retrieved successfully")
     @get_app_model
     @setup_required
@@ -93,12 +82,17 @@ class WorkflowDailyTerminalsStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        )
+        args = parser.parse_args()
 
         assert account.timezone is not None
 
         try:
-            start_date, end_date = parse_time_range(args.start, args.end, account.timezone)
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -124,7 +118,9 @@ class WorkflowDailyTokenCostStatistic(Resource):
     @console_ns.doc("get_workflow_daily_token_cost_statistic")
     @console_ns.doc(description="Get workflow daily token cost statistics")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[WorkflowStatisticQuery.__name__])
+    @console_ns.doc(
+        params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"}
+    )
     @console_ns.response(200, "Daily token cost statistics retrieved successfully")
     @get_app_model
     @setup_required
@@ -133,12 +129,17 @@ class WorkflowDailyTokenCostStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        )
+        args = parser.parse_args()
 
         assert account.timezone is not None
 
         try:
-            start_date, end_date = parse_time_range(args.start, args.end, account.timezone)
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 
@@ -164,7 +165,9 @@ class WorkflowAverageAppInteractionStatistic(Resource):
     @console_ns.doc("get_workflow_average_app_interaction_statistic")
     @console_ns.doc(description="Get workflow average app interaction statistics")
     @console_ns.doc(params={"app_id": "Application ID"})
-    @console_ns.expect(console_ns.models[WorkflowStatisticQuery.__name__])
+    @console_ns.doc(
+        params={"start": "Start date and time (YYYY-MM-DD HH:MM)", "end": "End date and time (YYYY-MM-DD HH:MM)"}
+    )
     @console_ns.response(200, "Average app interaction statistics retrieved successfully")
     @setup_required
     @login_required
@@ -173,12 +176,17 @@ class WorkflowAverageAppInteractionStatistic(Resource):
     def get(self, app_model):
         account, _ = current_account_with_tenant()
 
-        args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("start", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+            .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
+        )
+        args = parser.parse_args()
 
         assert account.timezone is not None
 
         try:
-            start_date, end_date = parse_time_range(args.start, args.end, account.timezone)
+            start_date, end_date = parse_time_range(args["start"], args["end"], account.timezone)
         except ValueError as e:
             abort(400, description=str(e))
 

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

@@ -1004,11 +1004,6 @@ class RagPipelineRecommendedPluginApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument('type', type=str, location='args', required=False, default='all')
-        args = parser.parse_args()
-        type = args["type"]
-
         rag_pipeline_service = RagPipelineService()
-        recommended_plugins = rag_pipeline_service.get_recommended_plugins(type)
+        recommended_plugins = rag_pipeline_service.get_recommended_plugins()
         return recommended_plugins

api/controllers/console/version.py

@@ -58,7 +58,7 @@ class VersionApi(Resource):
             response = httpx.get(
                 check_update_url,
                 params={"current_version": args["current_version"]},
-                timeout=httpx.Timeout(timeout=10.0, connect=3.0),
+                timeout=httpx.Timeout(connect=3, read=10),
             )
         except Exception as error:
             logger.warning("Check update version error: %s.", str(error))

api/controllers/console/workspace/account.py

@@ -174,25 +174,63 @@ class CheckEmailUniquePayload(BaseModel):
         return email(value)
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(AccountInitPayload)
-reg(AccountNamePayload)
-reg(AccountAvatarPayload)
-reg(AccountInterfaceLanguagePayload)
-reg(AccountInterfaceThemePayload)
-reg(AccountTimezonePayload)
-reg(AccountPasswordPayload)
-reg(AccountDeletePayload)
-reg(AccountDeletionFeedbackPayload)
-reg(EducationActivatePayload)
-reg(EducationAutocompleteQuery)
-reg(ChangeEmailSendPayload)
-reg(ChangeEmailValidityPayload)
-reg(ChangeEmailResetPayload)
-reg(CheckEmailUniquePayload)
+console_ns.schema_model(
+    AccountInitPayload.__name__, AccountInitPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    AccountNamePayload.__name__, AccountNamePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    AccountAvatarPayload.__name__, AccountAvatarPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+console_ns.schema_model(
+    AccountInterfaceLanguagePayload.__name__,
+    AccountInterfaceLanguagePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    AccountInterfaceThemePayload.__name__,
+    AccountInterfaceThemePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    AccountTimezonePayload.__name__,
+    AccountTimezonePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    AccountPasswordPayload.__name__,
+    AccountPasswordPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    AccountDeletePayload.__name__,
+    AccountDeletePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    AccountDeletionFeedbackPayload.__name__,
+    AccountDeletionFeedbackPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    EducationActivatePayload.__name__,
+    EducationActivatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    EducationAutocompleteQuery.__name__,
+    EducationAutocompleteQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    ChangeEmailSendPayload.__name__,
+    ChangeEmailSendPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    ChangeEmailValidityPayload.__name__,
+    ChangeEmailValidityPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    ChangeEmailResetPayload.__name__,
+    ChangeEmailResetPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    CheckEmailUniquePayload.__name__,
+    CheckEmailUniquePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
 
 @console_ns.route("/account/init")

api/controllers/console/workspace/endpoint.py

@@ -1,8 +1,4 @@
-from typing import Any
-
-from flask import request
-from flask_restx import Resource, fields
-from pydantic import BaseModel, Field
+from flask_restx import Resource, fields, reqparse
 
 from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, is_admin_or_owner_required, setup_required
@@ -11,49 +7,21 @@ from core.plugin.impl.exc import PluginPermissionDeniedError
 from libs.login import current_account_with_tenant, login_required
 from services.plugin.endpoint_service import EndpointService
 
-DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
-
-
-class EndpointCreatePayload(BaseModel):
-    plugin_unique_identifier: str
-    settings: dict[str, Any]
-    name: str = Field(min_length=1)
-
-
-class EndpointIdPayload(BaseModel):
-    endpoint_id: str
-
-
-class EndpointUpdatePayload(EndpointIdPayload):
-    settings: dict[str, Any]
-    name: str = Field(min_length=1)
-
-
-class EndpointListQuery(BaseModel):
-    page: int = Field(ge=1)
-    page_size: int = Field(gt=0)
-
-
-class EndpointListForPluginQuery(EndpointListQuery):
-    plugin_id: str
-
-
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(EndpointCreatePayload)
-reg(EndpointIdPayload)
-reg(EndpointUpdatePayload)
-reg(EndpointListQuery)
-reg(EndpointListForPluginQuery)
-
 
 @console_ns.route("/workspaces/current/endpoints/create")
 class EndpointCreateApi(Resource):
     @console_ns.doc("create_endpoint")
     @console_ns.doc(description="Create a new plugin endpoint")
-    @console_ns.expect(console_ns.models[EndpointCreatePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "EndpointCreateRequest",
+            {
+                "plugin_unique_identifier": fields.String(required=True, description="Plugin unique identifier"),
+                "settings": fields.Raw(required=True, description="Endpoint settings"),
+                "name": fields.String(required=True, description="Endpoint name"),
+            },
+        )
+    )
     @console_ns.response(
         200,
         "Endpoint created successfully",
@@ -67,16 +35,26 @@ class EndpointCreateApi(Resource):
     def post(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointCreatePayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("plugin_unique_identifier", type=str, required=True)
+            .add_argument("settings", type=dict, required=True)
+            .add_argument("name", type=str, required=True)
+        )
+        args = parser.parse_args()
+
+        plugin_unique_identifier = args["plugin_unique_identifier"]
+        settings = args["settings"]
+        name = args["name"]
 
         try:
             return {
                 "success": EndpointService.create_endpoint(
                     tenant_id=tenant_id,
                     user_id=user.id,
-                    plugin_unique_identifier=args.plugin_unique_identifier,
-                    name=args.name,
-                    settings=args.settings,
+                    plugin_unique_identifier=plugin_unique_identifier,
+                    name=name,
+                    settings=settings,
                 )
             }
         except PluginPermissionDeniedError as e:
@@ -87,7 +65,11 @@ class EndpointCreateApi(Resource):
 class EndpointListApi(Resource):
     @console_ns.doc("list_endpoints")
     @console_ns.doc(description="List plugin endpoints with pagination")
-    @console_ns.expect(console_ns.models[EndpointListQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("page", type=int, required=True, location="args", help="Page number")
+        .add_argument("page_size", type=int, required=True, location="args", help="Page size")
+    )
     @console_ns.response(
         200,
         "Success",
@@ -101,10 +83,15 @@ class EndpointListApi(Resource):
     def get(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointListQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("page", type=int, required=True, location="args")
+            .add_argument("page_size", type=int, required=True, location="args")
+        )
+        args = parser.parse_args()
 
-        page = args.page
-        page_size = args.page_size
+        page = args["page"]
+        page_size = args["page_size"]
 
         return jsonable_encoder(
             {
@@ -122,7 +109,12 @@ class EndpointListApi(Resource):
 class EndpointListForSinglePluginApi(Resource):
     @console_ns.doc("list_plugin_endpoints")
     @console_ns.doc(description="List endpoints for a specific plugin")
-    @console_ns.expect(console_ns.models[EndpointListForPluginQuery.__name__])
+    @console_ns.expect(
+        console_ns.parser()
+        .add_argument("page", type=int, required=True, location="args", help="Page number")
+        .add_argument("page_size", type=int, required=True, location="args", help="Page size")
+        .add_argument("plugin_id", type=str, required=True, location="args", help="Plugin ID")
+    )
     @console_ns.response(
         200,
         "Success",
@@ -136,11 +128,17 @@ class EndpointListForSinglePluginApi(Resource):
     def get(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointListForPluginQuery.model_validate(request.args.to_dict(flat=True))  # type: ignore
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("page", type=int, required=True, location="args")
+            .add_argument("page_size", type=int, required=True, location="args")
+            .add_argument("plugin_id", type=str, required=True, location="args")
+        )
+        args = parser.parse_args()
 
-        page = args.page
-        page_size = args.page_size
-        plugin_id = args.plugin_id
+        page = args["page"]
+        page_size = args["page_size"]
+        plugin_id = args["plugin_id"]
 
         return jsonable_encoder(
             {
@@ -159,7 +157,11 @@ class EndpointListForSinglePluginApi(Resource):
 class EndpointDeleteApi(Resource):
     @console_ns.doc("delete_endpoint")
     @console_ns.doc(description="Delete a plugin endpoint")
-    @console_ns.expect(console_ns.models[EndpointIdPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "EndpointDeleteRequest", {"endpoint_id": fields.String(required=True, description="Endpoint ID")}
+        )
+    )
     @console_ns.response(
         200,
         "Endpoint deleted successfully",
@@ -173,12 +175,13 @@ class EndpointDeleteApi(Resource):
     def post(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointIdPayload.model_validate(console_ns.payload)
+        parser = reqparse.RequestParser().add_argument("endpoint_id", type=str, required=True)
+        args = parser.parse_args()
+
+        endpoint_id = args["endpoint_id"]
 
         return {
-            "success": EndpointService.delete_endpoint(
-                tenant_id=tenant_id, user_id=user.id, endpoint_id=args.endpoint_id
-            )
+            "success": EndpointService.delete_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
         }
 
 
@@ -186,7 +189,16 @@ class EndpointDeleteApi(Resource):
 class EndpointUpdateApi(Resource):
     @console_ns.doc("update_endpoint")
     @console_ns.doc(description="Update a plugin endpoint")
-    @console_ns.expect(console_ns.models[EndpointUpdatePayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "EndpointUpdateRequest",
+            {
+                "endpoint_id": fields.String(required=True, description="Endpoint ID"),
+                "settings": fields.Raw(required=True, description="Updated settings"),
+                "name": fields.String(required=True, description="Updated name"),
+            },
+        )
+    )
     @console_ns.response(
         200,
         "Endpoint updated successfully",
@@ -200,15 +212,25 @@ class EndpointUpdateApi(Resource):
     def post(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointUpdatePayload.model_validate(console_ns.payload)
+        parser = (
+            reqparse.RequestParser()
+            .add_argument("endpoint_id", type=str, required=True)
+            .add_argument("settings", type=dict, required=True)
+            .add_argument("name", type=str, required=True)
+        )
+        args = parser.parse_args()
+
+        endpoint_id = args["endpoint_id"]
+        settings = args["settings"]
+        name = args["name"]
 
         return {
             "success": EndpointService.update_endpoint(
                 tenant_id=tenant_id,
                 user_id=user.id,
-                endpoint_id=args.endpoint_id,
-                name=args.name,
-                settings=args.settings,
+                endpoint_id=endpoint_id,
+                name=name,
+                settings=settings,
             )
         }
 
@@ -217,7 +239,11 @@ class EndpointUpdateApi(Resource):
 class EndpointEnableApi(Resource):
     @console_ns.doc("enable_endpoint")
     @console_ns.doc(description="Enable a plugin endpoint")
-    @console_ns.expect(console_ns.models[EndpointIdPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "EndpointEnableRequest", {"endpoint_id": fields.String(required=True, description="Endpoint ID")}
+        )
+    )
     @console_ns.response(
         200,
         "Endpoint enabled successfully",
@@ -231,12 +257,13 @@ class EndpointEnableApi(Resource):
     def post(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointIdPayload.model_validate(console_ns.payload)
+        parser = reqparse.RequestParser().add_argument("endpoint_id", type=str, required=True)
+        args = parser.parse_args()
+
+        endpoint_id = args["endpoint_id"]
 
         return {
-            "success": EndpointService.enable_endpoint(
-                tenant_id=tenant_id, user_id=user.id, endpoint_id=args.endpoint_id
-            )
+            "success": EndpointService.enable_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
         }
 
 
@@ -244,7 +271,11 @@ class EndpointEnableApi(Resource):
 class EndpointDisableApi(Resource):
     @console_ns.doc("disable_endpoint")
     @console_ns.doc(description="Disable a plugin endpoint")
-    @console_ns.expect(console_ns.models[EndpointIdPayload.__name__])
+    @console_ns.expect(
+        console_ns.model(
+            "EndpointDisableRequest", {"endpoint_id": fields.String(required=True, description="Endpoint ID")}
+        )
+    )
     @console_ns.response(
         200,
         "Endpoint disabled successfully",
@@ -258,10 +289,11 @@ class EndpointDisableApi(Resource):
     def post(self):
         user, tenant_id = current_account_with_tenant()
 
-        args = EndpointIdPayload.model_validate(console_ns.payload)
+        parser = reqparse.RequestParser().add_argument("endpoint_id", type=str, required=True)
+        args = parser.parse_args()
+
+        endpoint_id = args["endpoint_id"]
 
         return {
-            "success": EndpointService.disable_endpoint(
-                tenant_id=tenant_id, user_id=user.id, endpoint_id=args.endpoint_id
-            )
+            "success": EndpointService.disable_endpoint(tenant_id=tenant_id, user_id=user.id, endpoint_id=endpoint_id)
         }

api/controllers/console/workspace/members.py

@@ -58,15 +58,26 @@ class OwnerTransferPayload(BaseModel):
     token: str
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
-reg(MemberInvitePayload)
-reg(MemberRoleUpdatePayload)
-reg(OwnerTransferEmailPayload)
-reg(OwnerTransferCheckPayload)
-reg(OwnerTransferPayload)
+console_ns.schema_model(
+    MemberInvitePayload.__name__,
+    MemberInvitePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    MemberRoleUpdatePayload.__name__,
+    MemberRoleUpdatePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    OwnerTransferEmailPayload.__name__,
+    OwnerTransferEmailPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    OwnerTransferCheckPayload.__name__,
+    OwnerTransferCheckPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+console_ns.schema_model(
+    OwnerTransferPayload.__name__,
+    OwnerTransferPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
 
 @console_ns.route("/workspaces/current/members")

api/controllers/console/workspace/model_providers.py

@@ -75,18 +75,44 @@ class ParserPreferredProviderType(BaseModel):
     preferred_provider_type: Literal["system", "custom"]
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+console_ns.schema_model(
+    ParserModelList.__name__, ParserModelList.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
 
+console_ns.schema_model(
+    ParserCredentialId.__name__,
+    ParserCredentialId.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
-reg(ParserModelList)
-reg(ParserCredentialId)
-reg(ParserCredentialCreate)
-reg(ParserCredentialUpdate)
-reg(ParserCredentialDelete)
-reg(ParserCredentialSwitch)
-reg(ParserCredentialValidate)
-reg(ParserPreferredProviderType)
+console_ns.schema_model(
+    ParserCredentialCreate.__name__,
+    ParserCredentialCreate.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserCredentialUpdate.__name__,
+    ParserCredentialUpdate.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserCredentialDelete.__name__,
+    ParserCredentialDelete.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserCredentialSwitch.__name__,
+    ParserCredentialSwitch.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserCredentialValidate.__name__,
+    ParserCredentialValidate.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserPreferredProviderType.__name__,
+    ParserPreferredProviderType.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
 
 @console_ns.route("/workspaces/current/model-providers")

api/controllers/console/workspace/models.py

@@ -32,11 +32,25 @@ class ParserPostDefault(BaseModel):
     model_settings: list[Inner]
 
 
+console_ns.schema_model(
+    ParserGetDefault.__name__, ParserGetDefault.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserPostDefault.__name__, ParserPostDefault.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+
 class ParserDeleteModels(BaseModel):
     model: str
     model_type: ModelType
 
 
+console_ns.schema_model(
+    ParserDeleteModels.__name__, ParserDeleteModels.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+
 class LoadBalancingPayload(BaseModel):
     configs: list[dict[str, Any]] | None = None
     enabled: bool | None = None
@@ -105,19 +119,33 @@ class ParserParameter(BaseModel):
     model: str
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+console_ns.schema_model(
+    ParserPostModels.__name__, ParserPostModels.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
 
+console_ns.schema_model(
+    ParserGetCredentials.__name__,
+    ParserGetCredentials.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
-reg(ParserGetDefault)
-reg(ParserPostDefault)
-reg(ParserDeleteModels)
-reg(ParserPostModels)
-reg(ParserGetCredentials)
-reg(ParserCreateCredential)
-reg(ParserUpdateCredential)
-reg(ParserDeleteCredential)
-reg(ParserParameter)
+console_ns.schema_model(
+    ParserCreateCredential.__name__,
+    ParserCreateCredential.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserUpdateCredential.__name__,
+    ParserUpdateCredential.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserDeleteCredential.__name__,
+    ParserDeleteCredential.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserParameter.__name__, ParserParameter.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
 
 
 @console_ns.route("/workspaces/current/default-model")

api/controllers/console/workspace/plugin.py

@@ -22,10 +22,6 @@ from services.plugin.plugin_service import PluginService
 DEFAULT_REF_TEMPLATE_SWAGGER_2_0 = "#/definitions/{model}"
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
-
-
 @console_ns.route("/workspaces/current/plugin/debugging-key")
 class PluginDebuggingKeyApi(Resource):
     @setup_required
@@ -50,7 +46,9 @@ class ParserList(BaseModel):
     page_size: int = Field(default=256)
 
 
-reg(ParserList)
+console_ns.schema_model(
+    ParserList.__name__, ParserList.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
 
 
 @console_ns.route("/workspaces/current/plugin/list")
@@ -74,6 +72,11 @@ class ParserLatest(BaseModel):
     plugin_ids: list[str]
 
 
+console_ns.schema_model(
+    ParserLatest.__name__, ParserLatest.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+
 class ParserIcon(BaseModel):
     tenant_id: str
     filename: str
@@ -170,22 +173,72 @@ class ParserReadme(BaseModel):
     language: str = Field(default="en-US")
 
 
-reg(ParserLatest)
-reg(ParserIcon)
-reg(ParserAsset)
-reg(ParserGithubUpload)
-reg(ParserPluginIdentifiers)
-reg(ParserGithubInstall)
-reg(ParserPluginIdentifierQuery)
-reg(ParserTasks)
-reg(ParserMarketplaceUpgrade)
-reg(ParserGithubUpgrade)
-reg(ParserUninstall)
-reg(ParserPermissionChange)
-reg(ParserDynamicOptions)
-reg(ParserPreferencesChange)
-reg(ParserExcludePlugin)
-reg(ParserReadme)
+console_ns.schema_model(
+    ParserIcon.__name__, ParserIcon.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserAsset.__name__, ParserAsset.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserGithubUpload.__name__, ParserGithubUpload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserPluginIdentifiers.__name__,
+    ParserPluginIdentifiers.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserGithubInstall.__name__, ParserGithubInstall.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserPluginIdentifierQuery.__name__,
+    ParserPluginIdentifierQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserTasks.__name__, ParserTasks.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserMarketplaceUpgrade.__name__,
+    ParserMarketplaceUpgrade.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserGithubUpgrade.__name__, ParserGithubUpgrade.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserUninstall.__name__, ParserUninstall.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
+
+console_ns.schema_model(
+    ParserPermissionChange.__name__,
+    ParserPermissionChange.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserDynamicOptions.__name__,
+    ParserDynamicOptions.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserPreferencesChange.__name__,
+    ParserPreferencesChange.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserExcludePlugin.__name__,
+    ParserExcludePlugin.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    ParserReadme.__name__, ParserReadme.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
 
 
 @console_ns.route("/workspaces/current/plugin/list/latest-versions")

api/controllers/console/workspace/workspace.py

@@ -54,14 +54,25 @@ class WorkspaceInfoPayload(BaseModel):
     name: str
 
 
-def reg(cls: type[BaseModel]):
-    console_ns.schema_model(cls.__name__, cls.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0))
+console_ns.schema_model(
+    WorkspaceListQuery.__name__, WorkspaceListQuery.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0)
+)
 
+console_ns.schema_model(
+    SwitchWorkspacePayload.__name__,
+    SwitchWorkspacePayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    WorkspaceCustomConfigPayload.__name__,
+    WorkspaceCustomConfigPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
+console_ns.schema_model(
+    WorkspaceInfoPayload.__name__,
+    WorkspaceInfoPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
 
-reg(WorkspaceListQuery)
-reg(SwitchWorkspacePayload)
-reg(WorkspaceCustomConfigPayload)
-reg(WorkspaceInfoPayload)
 
 provider_fields = {
     "provider_name": fields.String,

api/controllers/web/__init__.py

@@ -20,6 +20,7 @@ from . import (
     audio,
     completion,
     conversation,
+    enduser_auth,
     feature,
     files,
     forgot_password,
@@ -41,6 +42,7 @@ __all__ = [
     "bp",
     "completion",
     "conversation",
+    "enduser_auth",
     "feature",
     "files",
     "forgot_password",

api/core/app/entities/app_invoke_entities.py

@@ -4,15 +4,15 @@ from typing import TYPE_CHECKING, Any, Optional
 
 from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator
 
+if TYPE_CHECKING:
+    from core.ops.ops_trace_manager import TraceQueueManager
+
 from constants import UUID_NIL
 from core.app.app_config.entities import EasyUIBasedAppConfig, WorkflowUIBasedAppConfig
 from core.entities.provider_configuration import ProviderModelBundle
 from core.file import File, FileUploadConfig
 from core.model_runtime.entities.model_entities import AIModelEntity
 
-if TYPE_CHECKING:
-    from core.ops.ops_trace_manager import TraceQueueManager
-
 
 class InvokeFrom(StrEnum):
     """
@@ -275,8 +275,10 @@ class RagPipelineGenerateEntity(WorkflowAppGenerateEntity):
     start_node_id: str | None = None
 
 
+# Import TraceQueueManager at runtime to resolve forward references
 from core.ops.ops_trace_manager import TraceQueueManager
 
+# Rebuild models that use forward references
 AppGenerateEntity.model_rebuild()
 EasyUIBasedAppGenerateEntity.model_rebuild()
 ConversationAppGenerateEntity.model_rebuild()

api/core/rag/datasource/vdb/oceanbase/oceanbase_vector.py

@@ -58,39 +58,11 @@ class OceanBaseVector(BaseVector):
             password=self._config.password,
             db_name=self._config.database,
         )
-        self._fields: list[str] = []  # List of fields in the collection
-        if self._client.check_table_exists(collection_name):
-            self._load_collection_fields()
         self._hybrid_search_enabled = self._check_hybrid_search_support()  # Check if hybrid search is supported
 
     def get_type(self) -> str:
         return VectorType.OCEANBASE
 
-    def _load_collection_fields(self):
-        """
-        Load collection fields from the database table.
-        This method populates the _fields list with column names from the table.
-        """
-        try:
-            if self._collection_name in self._client.metadata_obj.tables:
-                table = self._client.metadata_obj.tables[self._collection_name]
-                # Store all column names except 'id' (primary key)
-                self._fields = [column.name for column in table.columns if column.name != "id"]
-                logger.debug("Loaded fields for collection '%s': %s", self._collection_name, self._fields)
-            else:
-                logger.warning("Collection '%s' not found in metadata", self._collection_name)
-        except Exception as e:
-            logger.warning("Failed to load collection fields for '%s': %s", self._collection_name, str(e))
-
-    def field_exists(self, field: str) -> bool:
-        """
-        Check if a field exists in the collection.
-
-        :param field: Field name to check
-        :return: True if field exists, False otherwise
-        """
-        return field in self._fields
-
     def create(self, texts: list[Document], embeddings: list[list[float]], **kwargs):
         self._vec_dim = len(embeddings[0])
         self._create_collection()
@@ -179,7 +151,6 @@ class OceanBaseVector(BaseVector):
                 logger.debug("DEBUG: Hybrid search is NOT enabled for '%s'", self._collection_name)
 
             self._client.refresh_metadata([self._collection_name])
-            self._load_collection_fields()
             redis_client.set(collection_exist_cache_key, 1, ex=3600)
 
     def _check_hybrid_search_support(self) -> bool:
@@ -206,134 +177,42 @@ class OceanBaseVector(BaseVector):
     def add_texts(self, documents: list[Document], embeddings: list[list[float]], **kwargs):
         ids = self._get_uuids(documents)
         for id, doc, emb in zip(ids, documents, embeddings):
-            try:
-                self._client.insert(
-                    table_name=self._collection_name,
-                    data={
-                        "id": id,
-                        "vector": emb,
-                        "text": doc.page_content,
-                        "metadata": doc.metadata,
-                    },
-                )
-            except Exception as e:
-                logger.exception(
-                    "Failed to insert document with id '%s' in collection '%s'",
-                    id,
-                    self._collection_name,
-                )
-                raise Exception(f"Failed to insert document with id '{id}'") from e
+            self._client.insert(
+                table_name=self._collection_name,
+                data={
+                    "id": id,
+                    "vector": emb,
+                    "text": doc.page_content,
+                    "metadata": doc.metadata,
+                },
+            )
 
     def text_exists(self, id: str) -> bool:
-        try:
-            cur = self._client.get(table_name=self._collection_name, ids=id)
-            return bool(cur.rowcount != 0)
-        except Exception as e:
-            logger.exception(
-                "Failed to check if text exists with id '%s' in collection '%s'",
-                id,
-                self._collection_name,
-            )
-            raise Exception(f"Failed to check text existence for id '{id}'") from e
+        cur = self._client.get(table_name=self._collection_name, ids=id)
+        return bool(cur.rowcount != 0)
 
     def delete_by_ids(self, ids: list[str]):
         if not ids:
             return
-        try:
-            self._client.delete(table_name=self._collection_name, ids=ids)
-            logger.debug("Deleted %d documents from collection '%s'", len(ids), self._collection_name)
-        except Exception as e:
-            logger.exception(
-                "Failed to delete %d documents from collection '%s'",
-                len(ids),
-                self._collection_name,
-            )
-            raise Exception(f"Failed to delete documents from collection '{self._collection_name}'") from e
+        self._client.delete(table_name=self._collection_name, ids=ids)
 
     def get_ids_by_metadata_field(self, key: str, value: str) -> list[str]:
-        try:
-            import re
+        from sqlalchemy import text
 
-            from sqlalchemy import text
-
-            # Validate key to prevent injection in JSON path
-            if not re.match(r"^[a-zA-Z0-9_.]+$", key):
-                raise ValueError(f"Invalid characters in metadata key: {key}")
-
-            # Use parameterized query to prevent SQL injection
-            sql = text(f"SELECT id FROM `{self._collection_name}` WHERE metadata->>'$.{key}' = :value")
-
-            with self._client.engine.connect() as conn:
-                result = conn.execute(sql, {"value": value})
-                ids = [row[0] for row in result]
-
-            logger.debug(
-                "Found %d documents with metadata field '%s'='%s' in collection '%s'",
-                len(ids),
-                key,
-                value,
-                self._collection_name,
-            )
-            return ids
-        except Exception as e:
-            logger.exception(
-                "Failed to get IDs by metadata field '%s'='%s' in collection '%s'",
-                key,
-                value,
-                self._collection_name,
-            )
-            raise Exception(f"Failed to query documents by metadata field '{key}'") from e
+        cur = self._client.get(
+            table_name=self._collection_name,
+            ids=None,
+            where_clause=[text(f"metadata->>'$.{key}' = '{value}'")],
+            output_column_name=["id"],
+        )
+        return [row[0] for row in cur]
 
     def delete_by_metadata_field(self, key: str, value: str):
         ids = self.get_ids_by_metadata_field(key, value)
-        if ids:
-            self.delete_by_ids(ids)
-        else:
-            logger.debug("No documents found to delete with metadata field '%s'='%s'", key, value)
-
-    def _process_search_results(
-        self, results: list[tuple], score_threshold: float = 0.0, score_key: str = "score"
-    ) -> list[Document]:
-        """
-        Common method to process search results
-
-        :param results: Search results as list of tuples (text, metadata, score)
-        :param score_threshold: Score threshold for filtering
-        :param score_key: Key name for score in metadata
-        :return: List of documents
-        """
-        docs = []
-        for row in results:
-            text, metadata_str, score = row[0], row[1], row[2]
-
-            # Parse metadata JSON
-            try:
-                metadata = json.loads(metadata_str) if isinstance(metadata_str, str) else metadata_str
-            except json.JSONDecodeError:
-                logger.warning("Invalid JSON metadata: %s", metadata_str)
-                metadata = {}
-
-            # Add score to metadata
-            metadata[score_key] = score
-
-            # Filter by score threshold
-            if score >= score_threshold:
-                docs.append(Document(page_content=text, metadata=metadata))
-
-        return docs
+        self.delete_by_ids(ids)
 
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
         if not self._hybrid_search_enabled:
-            logger.warning(
-                "Full-text search is disabled: set OCEANBASE_ENABLE_HYBRID_SEARCH=true (requires OceanBase >= 4.3.5.1)."
-            )
-            return []
-        if not self.field_exists("text"):
-            logger.warning(
-                "Full-text search unavailable: collection '%s' missing 'text' field; "
-                "recreate the collection after enabling OCEANBASE_ENABLE_HYBRID_SEARCH to add fulltext index.",
-                self._collection_name,
-            )
             return []
 
         try:
@@ -341,24 +220,13 @@ class OceanBaseVector(BaseVector):
             if not isinstance(top_k, int) or top_k <= 0:
                 raise ValueError("top_k must be a positive integer")
 
-            score_threshold = float(kwargs.get("score_threshold") or 0.0)
-
-            # Build parameterized query to prevent SQL injection
-            from sqlalchemy import text
-
             document_ids_filter = kwargs.get("document_ids_filter")
-            params = {"query": query}
             where_clause = ""
-
             if document_ids_filter:
-                # Create parameterized placeholders for document IDs
-                placeholders = ", ".join(f":doc_id_{i}" for i in range(len(document_ids_filter)))
-                where_clause = f" AND metadata->>'$.document_id' IN ({placeholders})"
-                # Add document IDs to parameters
-                for i, doc_id in enumerate(document_ids_filter):
-                    params[f"doc_id_{i}"] = doc_id
+                document_ids = ", ".join(f"'{id}'" for id in document_ids_filter)
+                where_clause = f" AND metadata->>'$.document_id' IN ({document_ids})"
 
-            full_sql = f"""SELECT text, metadata, MATCH (text) AGAINST (:query) AS score
+            full_sql = f"""SELECT metadata, text, MATCH (text) AGAINST (:query) AS score
             FROM {self._collection_name}
             WHERE MATCH (text) AGAINST (:query) > 0
             {where_clause}
@@ -367,45 +235,41 @@ class OceanBaseVector(BaseVector):
 
             with self._client.engine.connect() as conn:
                 with conn.begin():
-                    result = conn.execute(text(full_sql), params)
+                    from sqlalchemy import text
+
+                    result = conn.execute(text(full_sql), {"query": query})
                     rows = result.fetchall()
 
-                    return self._process_search_results(rows, score_threshold=score_threshold)
+                    docs = []
+                    for row in rows:
+                        metadata_str, _text, score = row
+                        try:
+                            metadata = json.loads(metadata_str)
+                        except json.JSONDecodeError:
+                            logger.warning("Invalid JSON metadata: %s", metadata_str)
+                            metadata = {}
+                        metadata["score"] = score
+                        docs.append(Document(page_content=_text, metadata=metadata))
+
+                    return docs
         except Exception as e:
-            logger.exception(
-                "Failed to perform full-text search on collection '%s' with query '%s'",
-                self._collection_name,
-                query,
-            )
-            raise Exception(f"Full-text search failed for collection '{self._collection_name}'") from e
+            logger.warning("Failed to fulltext search: %s.", str(e))
+            return []
 
     def search_by_vector(self, query_vector: list[float], **kwargs: Any) -> list[Document]:
-        from sqlalchemy import text
-
         document_ids_filter = kwargs.get("document_ids_filter")
         _where_clause = None
         if document_ids_filter:
-            # Validate document IDs to prevent SQL injection
-            # Document IDs should be alphanumeric with hyphens and underscores
-            import re
-
-            for doc_id in document_ids_filter:
-                if not isinstance(doc_id, str) or not re.match(r"^[a-zA-Z0-9_-]+$", doc_id):
-                    raise ValueError(f"Invalid document ID format: {doc_id}")
-
-            # Safe to use in query after validation
             document_ids = ", ".join(f"'{id}'" for id in document_ids_filter)
             where_clause = f"metadata->>'$.document_id' in ({document_ids})"
+            from sqlalchemy import text
+
             _where_clause = [text(where_clause)]
         ef_search = kwargs.get("ef_search", self._hnsw_ef_search)
         if ef_search != self._hnsw_ef_search:
             self._client.set_ob_hnsw_ef_search(ef_search)
             self._hnsw_ef_search = ef_search
         topk = kwargs.get("top_k", 10)
-        try:
-            score_threshold = float(val) if (val := kwargs.get("score_threshold")) is not None else 0.0
-        except (ValueError, TypeError) as e:
-            raise ValueError(f"Invalid score_threshold parameter: {e}") from e
         try:
             cur = self._client.ann_search(
                 table_name=self._collection_name,
@@ -418,27 +282,21 @@ class OceanBaseVector(BaseVector):
                 where_clause=_where_clause,
             )
         except Exception as e:
-            logger.exception(
-                "Failed to perform vector search on collection '%s'",
-                self._collection_name,
+            raise Exception("Failed to search by vector. ", e)
+        docs = []
+        for _text, metadata, distance in cur:
+            metadata = json.loads(metadata)
+            metadata["score"] = 1 - distance / math.sqrt(2)
+            docs.append(
+                Document(
+                    page_content=_text,
+                    metadata=metadata,
+                )
             )
-            raise Exception(f"Vector search failed for collection '{self._collection_name}'") from e
-
-        # Convert distance to score and prepare results for processing
-        results = []
-        for _text, metadata_str, distance in cur:
-            score = 1 - distance / math.sqrt(2)
-            results.append((_text, metadata_str, score))
-
-        return self._process_search_results(results, score_threshold=score_threshold)
+        return docs
 
     def delete(self):
-        try:
-            self._client.drop_table_if_exist(self._collection_name)
-            logger.debug("Dropped collection '%s'", self._collection_name)
-        except Exception as e:
-            logger.exception("Failed to delete collection '%s'", self._collection_name)
-            raise Exception(f"Failed to delete collection '{self._collection_name}'") from e
+        self._client.drop_table_if_exist(self._collection_name)
 
 
 class OceanBaseVectorFactory(AbstractVectorFactory):

api/core/tools/entities/api_entities.py

@@ -54,8 +54,6 @@ class ToolProviderApiEntity(BaseModel):
     configuration: MCPConfiguration | None = Field(
         default=None, description="The timeout and sse_read_timeout of the MCP tool"
     )
-    # Workflow
-    workflow_app_id: str | None = Field(default=None, description="The app id of the workflow tool")
 
     @field_validator("tools", mode="before")
     @classmethod
@@ -89,8 +87,6 @@ class ToolProviderApiEntity(BaseModel):
             optional_fields.update(self.optional_field("is_dynamic_registration", self.is_dynamic_registration))
             optional_fields.update(self.optional_field("masked_headers", self.masked_headers))
             optional_fields.update(self.optional_field("original_headers", self.original_headers))
-        elif self.type == ToolProviderType.WORKFLOW:
-            optional_fields.update(self.optional_field("workflow_app_id", self.workflow_app_id))
         return {
             "id": self.id,
             "author": self.author,

api/core/tools/entities/tool_entities.py

@@ -123,6 +123,16 @@ class ApiProviderAuthType(StrEnum):
         raise ValueError(f"invalid mode value '{value}', expected one of: {valid}")
 
 
+class ToolAuthType(StrEnum):
+    """
+    Enum class for tool authentication type.
+    Determines whether OAuth credentials are workspace-level or end-user-level.
+    """
+
+    WORKSPACE = "workspace"
+    END_USER = "end_user"
+
+
 class ToolInvokeMessage(BaseModel):
     class TextMessage(BaseModel):
         text: str

api/core/workflow/nodes/agent/agent_node.py

@@ -70,6 +70,7 @@ class AgentNode(Node[AgentNodeData]):
     """
 
     node_type = NodeType.AGENT
+    _node_data: AgentNodeData
 
     @classmethod
     def version(cls) -> str:
@@ -81,8 +82,8 @@ class AgentNode(Node[AgentNodeData]):
         try:
             strategy = get_plugin_agent_strategy(
                 tenant_id=self.tenant_id,
-                agent_strategy_provider_name=self.node_data.agent_strategy_provider_name,
-                agent_strategy_name=self.node_data.agent_strategy_name,
+                agent_strategy_provider_name=self._node_data.agent_strategy_provider_name,
+                agent_strategy_name=self._node_data.agent_strategy_name,
             )
         except Exception as e:
             yield StreamCompletedEvent(
@@ -100,13 +101,13 @@ class AgentNode(Node[AgentNodeData]):
         parameters = self._generate_agent_parameters(
             agent_parameters=agent_parameters,
             variable_pool=self.graph_runtime_state.variable_pool,
-            node_data=self.node_data,
+            node_data=self._node_data,
             strategy=strategy,
         )
         parameters_for_log = self._generate_agent_parameters(
             agent_parameters=agent_parameters,
             variable_pool=self.graph_runtime_state.variable_pool,
-            node_data=self.node_data,
+            node_data=self._node_data,
             for_log=True,
             strategy=strategy,
         )
@@ -139,7 +140,7 @@ class AgentNode(Node[AgentNodeData]):
                 messages=message_stream,
                 tool_info={
                     "icon": self.agent_strategy_icon,
-                    "agent_strategy": self.node_data.agent_strategy_name,
+                    "agent_strategy": self._node_data.agent_strategy_name,
                 },
                 parameters_for_log=parameters_for_log,
                 user_id=self.user_id,
@@ -386,7 +387,7 @@ class AgentNode(Node[AgentNodeData]):
             current_plugin = next(
                 plugin
                 for plugin in plugins
-                if f"{plugin.plugin_id}/{plugin.name}" == self.node_data.agent_strategy_provider_name
+                if f"{plugin.plugin_id}/{plugin.name}" == self._node_data.agent_strategy_provider_name
             )
             icon = current_plugin.declaration.icon
         except StopIteration:

api/core/workflow/nodes/answer/answer_node.py

@@ -14,12 +14,14 @@ class AnswerNode(Node[AnswerNodeData]):
     node_type = NodeType.ANSWER
     execution_type = NodeExecutionType.RESPONSE
 
+    _node_data: AnswerNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
 
     def _run(self) -> NodeRunResult:
-        segments = self.graph_runtime_state.variable_pool.convert_template(self.node_data.answer)
+        segments = self.graph_runtime_state.variable_pool.convert_template(self._node_data.answer)
         files = self._extract_files_from_segments(segments.value)
         return NodeRunResult(
             status=WorkflowNodeExecutionStatus.SUCCEEDED,
@@ -69,4 +71,4 @@ class AnswerNode(Node[AnswerNodeData]):
         Returns:
             Template instance for this Answer node
         """
-        return Template.from_answer_template(self.node_data.answer)
+        return Template.from_answer_template(self._node_data.answer)

api/core/workflow/nodes/base/node.py

@@ -1,11 +1,7 @@
-import importlib
 import logging
-import operator
-import pkgutil
 from abc import abstractmethod
 from collections.abc import Generator, Mapping, Sequence
 from functools import singledispatchmethod
-from types import MappingProxyType
 from typing import Any, ClassVar, Generic, TypeVar, cast, get_args, get_origin
 from uuid import uuid4
 
@@ -138,34 +134,6 @@ class Node(Generic[NodeDataT]):
 
         cls._node_data_type = node_data_type
 
-        # Skip base class itself
-        if cls is Node:
-            return
-        # Only register production node implementations defined under core.workflow.nodes.*
-        # This prevents test helper subclasses from polluting the global registry and
-        # accidentally overriding real node types (e.g., a test Answer node).
-        module_name = getattr(cls, "__module__", "")
-        # Only register concrete subclasses that define node_type and version()
-        node_type = cls.node_type
-        version = cls.version()
-        bucket = Node._registry.setdefault(node_type, {})
-        if module_name.startswith("core.workflow.nodes."):
-            # Production node definitions take precedence and may override
-            bucket[version] = cls  # type: ignore[index]
-        else:
-            # External/test subclasses may register but must not override production
-            bucket.setdefault(version, cls)  # type: ignore[index]
-        # Maintain a "latest" pointer preferring numeric versions; fallback to lexicographic
-        version_keys = [v for v in bucket if v != "latest"]
-        numeric_pairs: list[tuple[str, int]] = []
-        for v in version_keys:
-            numeric_pairs.append((v, int(v)))
-        if numeric_pairs:
-            latest_key = max(numeric_pairs, key=operator.itemgetter(1))[0]
-        else:
-            latest_key = max(version_keys) if version_keys else version
-        bucket["latest"] = bucket[latest_key]
-
     @classmethod
     def _extract_node_data_type_from_generic(cls) -> type[BaseNodeData] | None:
         """
@@ -197,9 +165,6 @@ class Node(Generic[NodeDataT]):
 
         return None
 
-    # Global registry populated via __init_subclass__
-    _registry: ClassVar[dict["NodeType", dict[str, type["Node"]]]] = {}
-
     def __init__(
         self,
         id: str,
@@ -275,23 +240,23 @@ class Node(Generic[NodeDataT]):
         from core.workflow.nodes.tool.tool_node import ToolNode
 
         if isinstance(self, ToolNode):
-            start_event.provider_id = getattr(self.node_data, "provider_id", "")
-            start_event.provider_type = getattr(self.node_data, "provider_type", "")
+            start_event.provider_id = getattr(self.get_base_node_data(), "provider_id", "")
+            start_event.provider_type = getattr(self.get_base_node_data(), "provider_type", "")
 
         from core.workflow.nodes.datasource.datasource_node import DatasourceNode
 
         if isinstance(self, DatasourceNode):
-            plugin_id = getattr(self.node_data, "plugin_id", "")
-            provider_name = getattr(self.node_data, "provider_name", "")
+            plugin_id = getattr(self.get_base_node_data(), "plugin_id", "")
+            provider_name = getattr(self.get_base_node_data(), "provider_name", "")
 
             start_event.provider_id = f"{plugin_id}/{provider_name}"
-            start_event.provider_type = getattr(self.node_data, "provider_type", "")
+            start_event.provider_type = getattr(self.get_base_node_data(), "provider_type", "")
 
         from core.workflow.nodes.trigger_plugin.trigger_event_node import TriggerEventNode
 
         if isinstance(self, TriggerEventNode):
-            start_event.provider_id = getattr(self.node_data, "provider_id", "")
-            start_event.provider_type = getattr(self.node_data, "provider_type", "")
+            start_event.provider_id = getattr(self.get_base_node_data(), "provider_id", "")
+            start_event.provider_type = getattr(self.get_base_node_data(), "provider_type", "")
 
         from typing import cast
 
@@ -300,7 +265,7 @@ class Node(Generic[NodeDataT]):
 
         if isinstance(self, AgentNode):
             start_event.agent_strategy = AgentNodeStrategyInit(
-                name=cast(AgentNodeData, self.node_data).agent_strategy_name,
+                name=cast(AgentNodeData, self.get_base_node_data()).agent_strategy_name,
                 icon=self.agent_strategy_icon,
             )
 
@@ -430,29 +395,6 @@ class Node(Generic[NodeDataT]):
         # in `api/core/workflow/nodes/__init__.py`.
         raise NotImplementedError("subclasses of BaseNode must implement `version` method.")
 
-    @classmethod
-    def get_node_type_classes_mapping(cls) -> Mapping["NodeType", Mapping[str, type["Node"]]]:
-        """Return mapping of NodeType -> {version -> Node subclass} using __init_subclass__ registry.
-
-        Import all modules under core.workflow.nodes so subclasses register themselves on import.
-        Then we return a readonly view of the registry to avoid accidental mutation.
-        """
-        # Import all node modules to ensure they are loaded (thus registered)
-        import core.workflow.nodes as _nodes_pkg
-
-        for _, _modname, _ in pkgutil.walk_packages(_nodes_pkg.__path__, _nodes_pkg.__name__ + "."):
-            # Avoid importing modules that depend on the registry to prevent circular imports
-            # e.g. node_factory imports node_mapping which builds the mapping here.
-            if _modname in {
-                "core.workflow.nodes.node_factory",
-                "core.workflow.nodes.node_mapping",
-            }:
-                continue
-            importlib.import_module(_modname)
-
-        # Return a readonly view so callers can't mutate the registry by accident
-        return {nt: MappingProxyType(ver_map) for nt, ver_map in cls._registry.items()}
-
     @property
     def retry(self) -> bool:
         return False
@@ -477,6 +419,10 @@ class Node(Generic[NodeDataT]):
         """Get the default values dictionary for this node."""
         return self._node_data.default_value_dict
 
+    def get_base_node_data(self) -> BaseNodeData:
+        """Get the BaseNodeData object for this node."""
+        return self._node_data
+
     # Public interface properties that delegate to abstract methods
     @property
     def error_strategy(self) -> ErrorStrategy | None:
@@ -602,7 +548,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             start_at=event.start_at,
             inputs=event.inputs,
             metadata=event.metadata,
@@ -615,7 +561,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             index=event.index,
             pre_loop_output=event.pre_loop_output,
         )
@@ -626,7 +572,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             start_at=event.start_at,
             inputs=event.inputs,
             outputs=event.outputs,
@@ -640,7 +586,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             start_at=event.start_at,
             inputs=event.inputs,
             outputs=event.outputs,
@@ -655,7 +601,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             start_at=event.start_at,
             inputs=event.inputs,
             metadata=event.metadata,
@@ -668,7 +614,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             index=event.index,
             pre_iteration_output=event.pre_iteration_output,
         )
@@ -679,7 +625,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             start_at=event.start_at,
             inputs=event.inputs,
             outputs=event.outputs,
@@ -693,7 +639,7 @@ class Node(Generic[NodeDataT]):
             id=self._node_execution_id,
             node_id=self._node_id,
             node_type=self.node_type,
-            node_title=self.node_data.title,
+            node_title=self.get_base_node_data().title,
             start_at=event.start_at,
             inputs=event.inputs,
             outputs=event.outputs,

api/core/workflow/nodes/code/code_node.py

@@ -24,6 +24,8 @@ from .exc import (
 class CodeNode(Node[CodeNodeData]):
     node_type = NodeType.CODE
 
+    _node_data: CodeNodeData
+
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
         """
@@ -46,12 +48,12 @@ class CodeNode(Node[CodeNodeData]):
 
     def _run(self) -> NodeRunResult:
         # Get code language
-        code_language = self.node_data.code_language
-        code = self.node_data.code
+        code_language = self._node_data.code_language
+        code = self._node_data.code
 
         # Get variables
         variables = {}
-        for variable_selector in self.node_data.variables:
+        for variable_selector in self._node_data.variables:
             variable_name = variable_selector.variable
             variable = self.graph_runtime_state.variable_pool.get(variable_selector.value_selector)
             if isinstance(variable, ArrayFileSegment):
@@ -67,7 +69,7 @@ class CodeNode(Node[CodeNodeData]):
             )
 
             # Transform result
-            result = self._transform_result(result=result, output_schema=self.node_data.outputs)
+            result = self._transform_result(result=result, output_schema=self._node_data.outputs)
         except (CodeExecutionError, CodeNodeError) as e:
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED, inputs=variables, error=str(e), error_type=type(e).__name__
@@ -404,7 +406,7 @@ class CodeNode(Node[CodeNodeData]):
 
     @property
     def retry(self) -> bool:
-        return self.node_data.retry_config.retry_enabled
+        return self._node_data.retry_config.retry_enabled
 
     @staticmethod
     def _convert_boolean_to_int(value: bool | int | float | None) -> int | float | None:

api/core/workflow/nodes/datasource/datasource_node.py

@@ -42,6 +42,7 @@ class DatasourceNode(Node[DatasourceNodeData]):
     Datasource Node
     """
 
+    _node_data: DatasourceNodeData
     node_type = NodeType.DATASOURCE
     execution_type = NodeExecutionType.ROOT
 
@@ -50,7 +51,7 @@ class DatasourceNode(Node[DatasourceNodeData]):
         Run the datasource node
         """
 
-        node_data = self.node_data
+        node_data = self._node_data
         variable_pool = self.graph_runtime_state.variable_pool
         datasource_type_segement = variable_pool.get(["sys", SystemVariableKey.DATASOURCE_TYPE])
         if not datasource_type_segement:

api/core/workflow/nodes/document_extractor/node.py

@@ -43,12 +43,14 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
 
     node_type = NodeType.DOCUMENT_EXTRACTOR
 
+    _node_data: DocumentExtractorNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
 
     def _run(self):
-        variable_selector = self.node_data.variable_selector
+        variable_selector = self._node_data.variable_selector
         variable = self.graph_runtime_state.variable_pool.get(variable_selector)
 
         if variable is None:

api/core/workflow/nodes/end/end_node.py

@@ -9,6 +9,8 @@ class EndNode(Node[EndNodeData]):
     node_type = NodeType.END
     execution_type = NodeExecutionType.RESPONSE
 
+    _node_data: EndNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
@@ -20,7 +22,7 @@ class EndNode(Node[EndNodeData]):
         This method runs after streaming is complete (if streaming was enabled).
         It collects all output variables and returns them.
         """
-        output_variables = self.node_data.outputs
+        output_variables = self._node_data.outputs
 
         outputs = {}
         for variable_selector in output_variables:
@@ -42,6 +44,6 @@ class EndNode(Node[EndNodeData]):
             Template instance for this End node
         """
         outputs_config = [
-            {"variable": output.variable, "value_selector": output.value_selector} for output in self.node_data.outputs
+            {"variable": output.variable, "value_selector": output.value_selector} for output in self._node_data.outputs
         ]
         return Template.from_end_outputs(outputs_config)

api/core/workflow/nodes/http_request/node.py

@@ -34,6 +34,8 @@ logger = logging.getLogger(__name__)
 class HttpRequestNode(Node[HttpRequestNodeData]):
     node_type = NodeType.HTTP_REQUEST
 
+    _node_data: HttpRequestNodeData
+
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
         return {
@@ -67,8 +69,8 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
         process_data = {}
         try:
             http_executor = Executor(
-                node_data=self.node_data,
-                timeout=self._get_request_timeout(self.node_data),
+                node_data=self._node_data,
+                timeout=self._get_request_timeout(self._node_data),
                 variable_pool=self.graph_runtime_state.variable_pool,
                 max_retries=0,
             )
@@ -223,4 +225,4 @@ class HttpRequestNode(Node[HttpRequestNodeData]):
 
     @property
     def retry(self) -> bool:
-        return self.node_data.retry_config.retry_enabled
+        return self._node_data.retry_config.retry_enabled

api/core/workflow/nodes/human_input/human_input_node.py

@@ -25,6 +25,8 @@ class HumanInputNode(Node[HumanInputNodeData]):
         "handle",
     )
 
+    _node_data: HumanInputNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
@@ -47,12 +49,12 @@ class HumanInputNode(Node[HumanInputNodeData]):
     def _is_completion_ready(self) -> bool:
         """Determine whether all required inputs are satisfied."""
 
-        if not self.node_data.required_variables:
+        if not self._node_data.required_variables:
             return False
 
         variable_pool = self.graph_runtime_state.variable_pool
 
-        for selector_str in self.node_data.required_variables:
+        for selector_str in self._node_data.required_variables:
             parts = selector_str.split(".")
             if len(parts) != 2:
                 return False
@@ -72,7 +74,7 @@ class HumanInputNode(Node[HumanInputNodeData]):
             if handle:
                 return handle
 
-        default_values = self.node_data.default_value_dict
+        default_values = self._node_data.default_value_dict
         for key in self._BRANCH_SELECTION_KEYS:
             handle = self._normalize_branch_value(default_values.get(key))
             if handle:

api/core/workflow/nodes/if_else/if_else_node.py

@@ -16,6 +16,8 @@ class IfElseNode(Node[IfElseNodeData]):
     node_type = NodeType.IF_ELSE
     execution_type = NodeExecutionType.BRANCH
 
+    _node_data: IfElseNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
@@ -35,8 +37,8 @@ class IfElseNode(Node[IfElseNodeData]):
         condition_processor = ConditionProcessor()
         try:
             # Check if the new cases structure is used
-            if self.node_data.cases:
-                for case in self.node_data.cases:
+            if self._node_data.cases:
+                for case in self._node_data.cases:
                     input_conditions, group_result, final_result = condition_processor.process_conditions(
                         variable_pool=self.graph_runtime_state.variable_pool,
                         conditions=case.conditions,
@@ -62,8 +64,8 @@ class IfElseNode(Node[IfElseNodeData]):
                 input_conditions, group_result, final_result = _should_not_use_old_function(  # pyright: ignore [reportDeprecated]
                     condition_processor=condition_processor,
                     variable_pool=self.graph_runtime_state.variable_pool,
-                    conditions=self.node_data.conditions or [],
-                    operator=self.node_data.logical_operator or "and",
+                    conditions=self._node_data.conditions or [],
+                    operator=self._node_data.logical_operator or "and",
                 )
 
                 selected_case_id = "true" if final_result else "false"

api/core/workflow/nodes/iteration/iteration_node.py

@@ -65,6 +65,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
 
     node_type = NodeType.ITERATION
     execution_type = NodeExecutionType.CONTAINER
+    _node_data: IterationNodeData
 
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
@@ -135,10 +136,10 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
             )
 
     def _get_iterator_variable(self) -> ArraySegment | NoneSegment:
-        variable = self.graph_runtime_state.variable_pool.get(self.node_data.iterator_selector)
+        variable = self.graph_runtime_state.variable_pool.get(self._node_data.iterator_selector)
 
         if not variable:
-            raise IteratorVariableNotFoundError(f"iterator variable {self.node_data.iterator_selector} not found")
+            raise IteratorVariableNotFoundError(f"iterator variable {self._node_data.iterator_selector} not found")
 
         if not isinstance(variable, ArraySegment) and not isinstance(variable, NoneSegment):
             raise InvalidIteratorValueError(f"invalid iterator value: {variable}, please provide a list.")
@@ -173,7 +174,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
         return cast(list[object], iterator_list_value)
 
     def _validate_start_node(self) -> None:
-        if not self.node_data.start_node_id:
+        if not self._node_data.start_node_id:
             raise StartNodeIdNotFoundError(f"field start_node_id in iteration {self._node_id} not found")
 
     def _execute_iterations(
@@ -183,7 +184,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
         iter_run_map: dict[str, float],
         usage_accumulator: list[LLMUsage],
     ) -> Generator[GraphNodeEventBase | NodeEventBase, None, None]:
-        if self.node_data.is_parallel:
+        if self._node_data.is_parallel:
             # Parallel mode execution
             yield from self._execute_parallel_iterations(
                 iterator_list_value=iterator_list_value,
@@ -230,7 +231,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
         outputs.extend([None] * len(iterator_list_value))
 
         # Determine the number of parallel workers
-        max_workers = min(self.node_data.parallel_nums, len(iterator_list_value))
+        max_workers = min(self._node_data.parallel_nums, len(iterator_list_value))
 
         with ThreadPoolExecutor(max_workers=max_workers) as executor:
             # Submit all iteration tasks
@@ -286,7 +287,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
 
                 except Exception as e:
                     # Handle errors based on error_handle_mode
-                    match self.node_data.error_handle_mode:
+                    match self._node_data.error_handle_mode:
                         case ErrorHandleMode.TERMINATED:
                             # Cancel remaining futures and re-raise
                             for f in future_to_index:
@@ -299,7 +300,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
                             outputs[index] = None  # Will be filtered later
 
         # Remove None values if in REMOVE_ABNORMAL_OUTPUT mode
-        if self.node_data.error_handle_mode == ErrorHandleMode.REMOVE_ABNORMAL_OUTPUT:
+        if self._node_data.error_handle_mode == ErrorHandleMode.REMOVE_ABNORMAL_OUTPUT:
             outputs[:] = [output for output in outputs if output is not None]
 
     def _execute_single_iteration_parallel(
@@ -388,7 +389,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
         If flatten_output is True (default), flattens the list if all elements are lists.
         """
         # If flatten_output is disabled, return outputs as-is
-        if not self.node_data.flatten_output:
+        if not self._node_data.flatten_output:
             return outputs
 
         if not outputs:
@@ -568,14 +569,14 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
                 self._append_iteration_info_to_event(event=event, iter_run_index=current_index)
                 yield event
             elif isinstance(event, (GraphRunSucceededEvent, GraphRunPartialSucceededEvent)):
-                result = variable_pool.get(self.node_data.output_selector)
+                result = variable_pool.get(self._node_data.output_selector)
                 if result is None:
                     outputs.append(None)
                 else:
                     outputs.append(result.to_object())
                 return
             elif isinstance(event, GraphRunFailedEvent):
-                match self.node_data.error_handle_mode:
+                match self._node_data.error_handle_mode:
                     case ErrorHandleMode.TERMINATED:
                         raise IterationNodeError(event.error)
                     case ErrorHandleMode.CONTINUE_ON_ERROR:
@@ -626,7 +627,7 @@ class IterationNode(LLMUsageTrackingMixin, Node[IterationNodeData]):
 
         # Initialize the iteration graph with the new node factory
         iteration_graph = Graph.init(
-            graph_config=self.graph_config, node_factory=node_factory, root_node_id=self.node_data.start_node_id
+            graph_config=self.graph_config, node_factory=node_factory, root_node_id=self._node_data.start_node_id
         )
 
         if not iteration_graph:

api/core/workflow/nodes/iteration/iteration_start_node.py

@@ -11,6 +11,8 @@ class IterationStartNode(Node[IterationStartNodeData]):
 
     node_type = NodeType.ITERATION_START
 
+    _node_data: IterationStartNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"

api/core/workflow/nodes/knowledge_index/knowledge_index_node.py

@@ -35,11 +35,12 @@ default_retrieval_model = {
 
 
 class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
+    _node_data: KnowledgeIndexNodeData
     node_type = NodeType.KNOWLEDGE_INDEX
     execution_type = NodeExecutionType.RESPONSE
 
     def _run(self) -> NodeRunResult:  # type: ignore
-        node_data = self.node_data
+        node_data = self._node_data
         variable_pool = self.graph_runtime_state.variable_pool
         dataset_id = variable_pool.get(["sys", SystemVariableKey.DATASET_ID])
         if not dataset_id:

api/core/workflow/nodes/knowledge_retrieval/knowledge_retrieval_node.py

@@ -83,6 +83,8 @@ default_retrieval_model = {
 class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeData]):
     node_type = NodeType.KNOWLEDGE_RETRIEVAL
 
+    _node_data: KnowledgeRetrievalNodeData
+
     # Instance attributes specific to LLMNode.
     # Output variable for file
     _file_outputs: list["File"]
@@ -120,7 +122,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
 
     def _run(self) -> NodeRunResult:
         # extract variables
-        variable = self.graph_runtime_state.variable_pool.get(self.node_data.query_variable_selector)
+        variable = self.graph_runtime_state.variable_pool.get(self._node_data.query_variable_selector)
         if not isinstance(variable, StringSegment):
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED,
@@ -161,7 +163,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
         # retrieve knowledge
         usage = LLMUsage.empty_usage()
         try:
-            results, usage = self._fetch_dataset_retriever(node_data=self.node_data, query=query)
+            results, usage = self._fetch_dataset_retriever(node_data=self._node_data, query=query)
             outputs = {"result": ArrayObjectSegment(value=results)}
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.SUCCEEDED,
@@ -534,7 +536,7 @@ class KnowledgeRetrievalNode(LLMUsageTrackingMixin, Node[KnowledgeRetrievalNodeD
                 prompt_messages=prompt_messages,
                 stop=stop,
                 user_id=self.user_id,
-                structured_output_enabled=self.node_data.structured_output_enabled,
+                structured_output_enabled=self._node_data.structured_output_enabled,
                 structured_output=None,
                 file_saver=self._llm_file_saver,
                 file_outputs=self._file_outputs,

api/core/workflow/nodes/list_operator/node.py

@@ -37,6 +37,8 @@ def _negation(filter_: Callable[[_T], bool]) -> Callable[[_T], bool]:
 class ListOperatorNode(Node[ListOperatorNodeData]):
     node_type = NodeType.LIST_OPERATOR
 
+    _node_data: ListOperatorNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
@@ -46,9 +48,9 @@ class ListOperatorNode(Node[ListOperatorNodeData]):
         process_data: dict[str, Sequence[object]] = {}
         outputs: dict[str, Any] = {}
 
-        variable = self.graph_runtime_state.variable_pool.get(self.node_data.variable)
+        variable = self.graph_runtime_state.variable_pool.get(self._node_data.variable)
         if variable is None:
-            error_message = f"Variable not found for selector: {self.node_data.variable}"
+            error_message = f"Variable not found for selector: {self._node_data.variable}"
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED, error=error_message, inputs=inputs, outputs=outputs
             )
@@ -67,7 +69,7 @@ class ListOperatorNode(Node[ListOperatorNodeData]):
                 outputs=outputs,
             )
         if not isinstance(variable, _SUPPORTED_TYPES_TUPLE):
-            error_message = f"Variable {self.node_data.variable} is not an array type, actual type: {type(variable)}"
+            error_message = f"Variable {self._node_data.variable} is not an array type, actual type: {type(variable)}"
             return NodeRunResult(
                 status=WorkflowNodeExecutionStatus.FAILED, error=error_message, inputs=inputs, outputs=outputs
             )
@@ -81,19 +83,19 @@ class ListOperatorNode(Node[ListOperatorNodeData]):
 
         try:
             # Filter
-            if self.node_data.filter_by.enabled:
+            if self._node_data.filter_by.enabled:
                 variable = self._apply_filter(variable)
 
             # Extract
-            if self.node_data.extract_by.enabled:
+            if self._node_data.extract_by.enabled:
                 variable = self._extract_slice(variable)
 
             # Order
-            if self.node_data.order_by.enabled:
+            if self._node_data.order_by.enabled:
                 variable = self._apply_order(variable)
 
             # Slice
-            if self.node_data.limit.enabled:
+            if self._node_data.limit.enabled:
                 variable = self._apply_slice(variable)
 
             outputs = {
@@ -119,7 +121,7 @@ class ListOperatorNode(Node[ListOperatorNodeData]):
     def _apply_filter(self, variable: _SUPPORTED_TYPES_ALIAS) -> _SUPPORTED_TYPES_ALIAS:
         filter_func: Callable[[Any], bool]
         result: list[Any] = []
-        for condition in self.node_data.filter_by.conditions:
+        for condition in self._node_data.filter_by.conditions:
             if isinstance(variable, ArrayStringSegment):
                 if not isinstance(condition.value, str):
                     raise InvalidFilterValueError(f"Invalid filter value: {condition.value}")
@@ -158,22 +160,22 @@ class ListOperatorNode(Node[ListOperatorNodeData]):
 
     def _apply_order(self, variable: _SUPPORTED_TYPES_ALIAS) -> _SUPPORTED_TYPES_ALIAS:
         if isinstance(variable, (ArrayStringSegment, ArrayNumberSegment, ArrayBooleanSegment)):
-            result = sorted(variable.value, reverse=self.node_data.order_by.value == Order.DESC)
+            result = sorted(variable.value, reverse=self._node_data.order_by.value == Order.DESC)
             variable = variable.model_copy(update={"value": result})
         else:
             result = _order_file(
-                order=self.node_data.order_by.value, order_by=self.node_data.order_by.key, array=variable.value
+                order=self._node_data.order_by.value, order_by=self._node_data.order_by.key, array=variable.value
             )
             variable = variable.model_copy(update={"value": result})
 
         return variable
 
     def _apply_slice(self, variable: _SUPPORTED_TYPES_ALIAS) -> _SUPPORTED_TYPES_ALIAS:
-        result = variable.value[: self.node_data.limit.size]
+        result = variable.value[: self._node_data.limit.size]
         return variable.model_copy(update={"value": result})
 
     def _extract_slice(self, variable: _SUPPORTED_TYPES_ALIAS) -> _SUPPORTED_TYPES_ALIAS:
-        value = int(self.graph_runtime_state.variable_pool.convert_template(self.node_data.extract_by.serial).text)
+        value = int(self.graph_runtime_state.variable_pool.convert_template(self._node_data.extract_by.serial).text)
         if value < 1:
             raise ValueError(f"Invalid serial index: must be >= 1, got {value}")
         if value > len(variable.value):

api/core/workflow/nodes/llm/node.py

@@ -102,6 +102,8 @@ logger = logging.getLogger(__name__)
 class LLMNode(Node[LLMNodeData]):
     node_type = NodeType.LLM
 
+    _node_data: LLMNodeData
+
     # Compiled regex for extracting <think> blocks (with compatibility for attributes)
     _THINK_PATTERN = re.compile(r"<think[^>]*>(.*?)</think>", re.IGNORECASE | re.DOTALL)
 
@@ -152,13 +154,13 @@ class LLMNode(Node[LLMNodeData]):
 
         try:
             # init messages template
-            self.node_data.prompt_template = self._transform_chat_messages(self.node_data.prompt_template)
+            self._node_data.prompt_template = self._transform_chat_messages(self._node_data.prompt_template)
 
             # fetch variables and fetch values from variable pool
-            inputs = self._fetch_inputs(node_data=self.node_data)
+            inputs = self._fetch_inputs(node_data=self._node_data)
 
             # fetch jinja2 inputs
-            jinja_inputs = self._fetch_jinja_inputs(node_data=self.node_data)
+            jinja_inputs = self._fetch_jinja_inputs(node_data=self._node_data)
 
             # merge inputs
             inputs.update(jinja_inputs)
@@ -167,9 +169,9 @@ class LLMNode(Node[LLMNodeData]):
             files = (
                 llm_utils.fetch_files(
                     variable_pool=variable_pool,
-                    selector=self.node_data.vision.configs.variable_selector,
+                    selector=self._node_data.vision.configs.variable_selector,
                 )
-                if self.node_data.vision.enabled
+                if self._node_data.vision.enabled
                 else []
             )
 
@@ -177,7 +179,7 @@ class LLMNode(Node[LLMNodeData]):
                 node_inputs["#files#"] = [file.to_dict() for file in files]
 
             # fetch context value
-            generator = self._fetch_context(node_data=self.node_data)
+            generator = self._fetch_context(node_data=self._node_data)
             context = None
             for event in generator:
                 context = event.context
@@ -187,7 +189,7 @@ class LLMNode(Node[LLMNodeData]):
 
             # fetch model config
             model_instance, model_config = LLMNode._fetch_model_config(
-                node_data_model=self.node_data.model,
+                node_data_model=self._node_data.model,
                 tenant_id=self.tenant_id,
             )
 
@@ -195,13 +197,13 @@ class LLMNode(Node[LLMNodeData]):
             memory = llm_utils.fetch_memory(
                 variable_pool=variable_pool,
                 app_id=self.app_id,
-                node_data_memory=self.node_data.memory,
+                node_data_memory=self._node_data.memory,
                 model_instance=model_instance,
             )
 
             query: str | None = None
-            if self.node_data.memory:
-                query = self.node_data.memory.query_prompt_template
+            if self._node_data.memory:
+                query = self._node_data.memory.query_prompt_template
                 if not query and (
                     query_variable := variable_pool.get((SYSTEM_VARIABLE_NODE_ID, SystemVariableKey.QUERY))
                 ):
@@ -213,29 +215,29 @@ class LLMNode(Node[LLMNodeData]):
                 context=context,
                 memory=memory,
                 model_config=model_config,
-                prompt_template=self.node_data.prompt_template,
-                memory_config=self.node_data.memory,
-                vision_enabled=self.node_data.vision.enabled,
-                vision_detail=self.node_data.vision.configs.detail,
+                prompt_template=self._node_data.prompt_template,
+                memory_config=self._node_data.memory,
+                vision_enabled=self._node_data.vision.enabled,
+                vision_detail=self._node_data.vision.configs.detail,
                 variable_pool=variable_pool,
-                jinja2_variables=self.node_data.prompt_config.jinja2_variables,
+                jinja2_variables=self._node_data.prompt_config.jinja2_variables,
                 tenant_id=self.tenant_id,
             )
 
             # handle invoke result
             generator = LLMNode.invoke_llm(
-                node_data_model=self.node_data.model,
+                node_data_model=self._node_data.model,
                 model_instance=model_instance,
                 prompt_messages=prompt_messages,
                 stop=stop,
                 user_id=self.user_id,
-                structured_output_enabled=self.node_data.structured_output_enabled,
-                structured_output=self.node_data.structured_output,
+                structured_output_enabled=self._node_data.structured_output_enabled,
+                structured_output=self._node_data.structured_output,
                 file_saver=self._llm_file_saver,
                 file_outputs=self._file_outputs,
                 node_id=self._node_id,
                 node_type=self.node_type,
-                reasoning_format=self.node_data.reasoning_format,
+                reasoning_format=self._node_data.reasoning_format,
             )
 
             structured_output: LLMStructuredOutput | None = None
@@ -251,12 +253,12 @@ class LLMNode(Node[LLMNodeData]):
                     reasoning_content = event.reasoning_content or ""
 
                     # For downstream nodes, determine clean text based on reasoning_format
-                    if self.node_data.reasoning_format == "tagged":
+                    if self._node_data.reasoning_format == "tagged":
                         # Keep <think> tags for backward compatibility
                         clean_text = result_text
                     else:
                         # Extract clean text from <think> tags
-                        clean_text, _ = LLMNode._split_reasoning(result_text, self.node_data.reasoning_format)
+                        clean_text, _ = LLMNode._split_reasoning(result_text, self._node_data.reasoning_format)
 
                     # Process structured output if available from the event.
                     structured_output = (
@@ -1202,7 +1204,7 @@ class LLMNode(Node[LLMNodeData]):
 
     @property
     def retry(self) -> bool:
-        return self.node_data.retry_config.retry_enabled
+        return self._node_data.retry_config.retry_enabled
 
 
 def _combine_message_content_with_role(

api/core/workflow/nodes/loop/loop_end_node.py

@@ -11,6 +11,8 @@ class LoopEndNode(Node[LoopEndNodeData]):
 
     node_type = NodeType.LOOP_END
 
+    _node_data: LoopEndNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"

api/core/workflow/nodes/loop/loop_node.py

@@ -46,6 +46,7 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
     """
 
     node_type = NodeType.LOOP
+    _node_data: LoopNodeData
     execution_type = NodeExecutionType.CONTAINER
 
     @classmethod
@@ -55,27 +56,27 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
     def _run(self) -> Generator:
         """Run the node."""
         # Get inputs
-        loop_count = self.node_data.loop_count
-        break_conditions = self.node_data.break_conditions
-        logical_operator = self.node_data.logical_operator
+        loop_count = self._node_data.loop_count
+        break_conditions = self._node_data.break_conditions
+        logical_operator = self._node_data.logical_operator
 
         inputs = {"loop_count": loop_count}
 
-        if not self.node_data.start_node_id:
+        if not self._node_data.start_node_id:
             raise ValueError(f"field start_node_id in loop {self._node_id} not found")
 
-        root_node_id = self.node_data.start_node_id
+        root_node_id = self._node_data.start_node_id
 
         # Initialize loop variables in the original variable pool
         loop_variable_selectors = {}
-        if self.node_data.loop_variables:
+        if self._node_data.loop_variables:
             value_processor: dict[Literal["constant", "variable"], Callable[[LoopVariableData], Segment | None]] = {
                 "constant": lambda var: self._get_segment_for_constant(var.var_type, var.value),
                 "variable": lambda var: self.graph_runtime_state.variable_pool.get(var.value)
                 if isinstance(var.value, list)
                 else None,
             }
-            for loop_variable in self.node_data.loop_variables:
+            for loop_variable in self._node_data.loop_variables:
                 if loop_variable.value_type not in value_processor:
                     raise ValueError(
                         f"Invalid value type '{loop_variable.value_type}' for loop variable {loop_variable.label}"
@@ -163,7 +164,7 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
 
                 yield LoopNextEvent(
                     index=i + 1,
-                    pre_loop_output=self.node_data.outputs,
+                    pre_loop_output=self._node_data.outputs,
                 )
 
             self._accumulate_usage(loop_usage)
@@ -171,7 +172,7 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
             yield LoopSucceededEvent(
                 start_at=start_at,
                 inputs=inputs,
-                outputs=self.node_data.outputs,
+                outputs=self._node_data.outputs,
                 steps=loop_count,
                 metadata={
                     WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS: loop_usage.total_tokens,
@@ -193,7 +194,7 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
                         WorkflowNodeExecutionMetadataKey.LOOP_DURATION_MAP: loop_duration_map,
                         WorkflowNodeExecutionMetadataKey.LOOP_VARIABLE_MAP: single_loop_variable_map,
                     },
-                    outputs=self.node_data.outputs,
+                    outputs=self._node_data.outputs,
                     inputs=inputs,
                     llm_usage=loop_usage,
                 )
@@ -251,11 +252,11 @@ class LoopNode(LLMUsageTrackingMixin, Node[LoopNodeData]):
             if isinstance(event, GraphRunFailedEvent):
                 raise Exception(event.error)
 
-        for loop_var in self.node_data.loop_variables or []:
+        for loop_var in self._node_data.loop_variables or []:
             key, sel = loop_var.label, [self._node_id, loop_var.label]
             segment = self.graph_runtime_state.variable_pool.get(sel)
-            self.node_data.outputs[key] = segment.value if segment else None
-        self.node_data.outputs["loop_round"] = current_index + 1
+            self._node_data.outputs[key] = segment.value if segment else None
+        self._node_data.outputs["loop_round"] = current_index + 1
 
         return reach_break_node
 

api/core/workflow/nodes/loop/loop_start_node.py

@@ -11,6 +11,8 @@ class LoopStartNode(Node[LoopStartNodeData]):
 
     node_type = NodeType.LOOP_START
 
+    _node_data: LoopStartNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"

api/core/workflow/nodes/node_mapping.py

@@ -1,9 +1,165 @@
 from collections.abc import Mapping
 
 from core.workflow.enums import NodeType
+from core.workflow.nodes.agent.agent_node import AgentNode
+from core.workflow.nodes.answer.answer_node import AnswerNode
 from core.workflow.nodes.base.node import Node
+from core.workflow.nodes.code import CodeNode
+from core.workflow.nodes.datasource.datasource_node import DatasourceNode
+from core.workflow.nodes.document_extractor import DocumentExtractorNode
+from core.workflow.nodes.end.end_node import EndNode
+from core.workflow.nodes.http_request import HttpRequestNode
+from core.workflow.nodes.human_input import HumanInputNode
+from core.workflow.nodes.if_else import IfElseNode
+from core.workflow.nodes.iteration import IterationNode, IterationStartNode
+from core.workflow.nodes.knowledge_index import KnowledgeIndexNode
+from core.workflow.nodes.knowledge_retrieval import KnowledgeRetrievalNode
+from core.workflow.nodes.list_operator import ListOperatorNode
+from core.workflow.nodes.llm import LLMNode
+from core.workflow.nodes.loop import LoopEndNode, LoopNode, LoopStartNode
+from core.workflow.nodes.parameter_extractor import ParameterExtractorNode
+from core.workflow.nodes.question_classifier import QuestionClassifierNode
+from core.workflow.nodes.start import StartNode
+from core.workflow.nodes.template_transform import TemplateTransformNode
+from core.workflow.nodes.tool import ToolNode
+from core.workflow.nodes.trigger_plugin import TriggerEventNode
+from core.workflow.nodes.trigger_schedule import TriggerScheduleNode
+from core.workflow.nodes.trigger_webhook import TriggerWebhookNode
+from core.workflow.nodes.variable_aggregator import VariableAggregatorNode
+from core.workflow.nodes.variable_assigner.v1 import VariableAssignerNode as VariableAssignerNodeV1
+from core.workflow.nodes.variable_assigner.v2 import VariableAssignerNode as VariableAssignerNodeV2
 
 LATEST_VERSION = "latest"
 
-# Mapping is built by Node.get_node_type_classes_mapping(), which imports and walks core.workflow.nodes
-NODE_TYPE_CLASSES_MAPPING: Mapping[NodeType, Mapping[str, type[Node]]] = Node.get_node_type_classes_mapping()
+# NOTE(QuantumGhost): This should be in sync with subclasses of BaseNode.
+# Specifically, if you have introduced new node types, you should add them here.
+#
+# TODO(QuantumGhost): This could be automated with either metaclass or `__init_subclass__`
+# hook. Try to avoid duplication of node information.
+NODE_TYPE_CLASSES_MAPPING: Mapping[NodeType, Mapping[str, type[Node]]] = {
+    NodeType.START: {
+        LATEST_VERSION: StartNode,
+        "1": StartNode,
+    },
+    NodeType.END: {
+        LATEST_VERSION: EndNode,
+        "1": EndNode,
+    },
+    NodeType.ANSWER: {
+        LATEST_VERSION: AnswerNode,
+        "1": AnswerNode,
+    },
+    NodeType.LLM: {
+        LATEST_VERSION: LLMNode,
+        "1": LLMNode,
+    },
+    NodeType.KNOWLEDGE_RETRIEVAL: {
+        LATEST_VERSION: KnowledgeRetrievalNode,
+        "1": KnowledgeRetrievalNode,
+    },
+    NodeType.IF_ELSE: {
+        LATEST_VERSION: IfElseNode,
+        "1": IfElseNode,
+    },
+    NodeType.CODE: {
+        LATEST_VERSION: CodeNode,
+        "1": CodeNode,
+    },
+    NodeType.TEMPLATE_TRANSFORM: {
+        LATEST_VERSION: TemplateTransformNode,
+        "1": TemplateTransformNode,
+    },
+    NodeType.QUESTION_CLASSIFIER: {
+        LATEST_VERSION: QuestionClassifierNode,
+        "1": QuestionClassifierNode,
+    },
+    NodeType.HTTP_REQUEST: {
+        LATEST_VERSION: HttpRequestNode,
+        "1": HttpRequestNode,
+    },
+    NodeType.TOOL: {
+        LATEST_VERSION: ToolNode,
+        # This is an issue that caused problems before.
+        # Logically, we shouldn't use two different versions to point to the same class here,
+        # but in order to maintain compatibility with historical data, this approach has been retained.
+        "2": ToolNode,
+        "1": ToolNode,
+    },
+    NodeType.VARIABLE_AGGREGATOR: {
+        LATEST_VERSION: VariableAggregatorNode,
+        "1": VariableAggregatorNode,
+    },
+    NodeType.LEGACY_VARIABLE_AGGREGATOR: {
+        LATEST_VERSION: VariableAggregatorNode,
+        "1": VariableAggregatorNode,
+    },  # original name of VARIABLE_AGGREGATOR
+    NodeType.ITERATION: {
+        LATEST_VERSION: IterationNode,
+        "1": IterationNode,
+    },
+    NodeType.ITERATION_START: {
+        LATEST_VERSION: IterationStartNode,
+        "1": IterationStartNode,
+    },
+    NodeType.LOOP: {
+        LATEST_VERSION: LoopNode,
+        "1": LoopNode,
+    },
+    NodeType.LOOP_START: {
+        LATEST_VERSION: LoopStartNode,
+        "1": LoopStartNode,
+    },
+    NodeType.LOOP_END: {
+        LATEST_VERSION: LoopEndNode,
+        "1": LoopEndNode,
+    },
+    NodeType.PARAMETER_EXTRACTOR: {
+        LATEST_VERSION: ParameterExtractorNode,
+        "1": ParameterExtractorNode,
+    },
+    NodeType.VARIABLE_ASSIGNER: {
+        LATEST_VERSION: VariableAssignerNodeV2,
+        "1": VariableAssignerNodeV1,
+        "2": VariableAssignerNodeV2,
+    },
+    NodeType.DOCUMENT_EXTRACTOR: {
+        LATEST_VERSION: DocumentExtractorNode,
+        "1": DocumentExtractorNode,
+    },
+    NodeType.LIST_OPERATOR: {
+        LATEST_VERSION: ListOperatorNode,
+        "1": ListOperatorNode,
+    },
+    NodeType.AGENT: {
+        LATEST_VERSION: AgentNode,
+        # This is an issue that caused problems before.
+        # Logically, we shouldn't use two different versions to point to the same class here,
+        # but in order to maintain compatibility with historical data, this approach has been retained.
+        "2": AgentNode,
+        "1": AgentNode,
+    },
+    NodeType.HUMAN_INPUT: {
+        LATEST_VERSION: HumanInputNode,
+        "1": HumanInputNode,
+    },
+    NodeType.DATASOURCE: {
+        LATEST_VERSION: DatasourceNode,
+        "1": DatasourceNode,
+    },
+    NodeType.KNOWLEDGE_INDEX: {
+        LATEST_VERSION: KnowledgeIndexNode,
+        "1": KnowledgeIndexNode,
+    },
+    NodeType.TRIGGER_WEBHOOK: {
+        LATEST_VERSION: TriggerWebhookNode,
+        "1": TriggerWebhookNode,
+    },
+    NodeType.TRIGGER_PLUGIN: {
+        LATEST_VERSION: TriggerEventNode,
+        "1": TriggerEventNode,
+    },
+    NodeType.TRIGGER_SCHEDULE: {
+        LATEST_VERSION: TriggerScheduleNode,
+        "1": TriggerScheduleNode,
+    },
+}

api/core/workflow/nodes/parameter_extractor/parameter_extractor_node.py

@@ -90,6 +90,8 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
 
     node_type = NodeType.PARAMETER_EXTRACTOR
 
+    _node_data: ParameterExtractorNodeData
+
     _model_instance: ModelInstance | None = None
     _model_config: ModelConfigWithCredentialsEntity | None = None
 
@@ -114,7 +116,7 @@ class ParameterExtractorNode(Node[ParameterExtractorNodeData]):
         """
         Run the node.
         """
-        node_data = self.node_data
+        node_data = self._node_data
         variable = self.graph_runtime_state.variable_pool.get(node_data.query)
         query = variable.text if variable else ""
 

api/core/workflow/nodes/question_classifier/question_classifier_node.py

@@ -47,6 +47,8 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
     node_type = NodeType.QUESTION_CLASSIFIER
     execution_type = NodeExecutionType.BRANCH
 
+    _node_data: QuestionClassifierNodeData
+
     _file_outputs: list["File"]
     _llm_file_saver: LLMFileSaver
 
@@ -80,7 +82,7 @@ class QuestionClassifierNode(Node[QuestionClassifierNodeData]):
         return "1"
 
     def _run(self):
-        node_data = self.node_data
+        node_data = self._node_data
         variable_pool = self.graph_runtime_state.variable_pool
 
         # extract variables

api/core/workflow/nodes/start/start_node.py

@@ -9,6 +9,8 @@ class StartNode(Node[StartNodeData]):
     node_type = NodeType.START
     execution_type = NodeExecutionType.ROOT
 
+    _node_data: StartNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"

api/core/workflow/nodes/template_transform/template_transform_node.py

@@ -14,6 +14,8 @@ MAX_TEMPLATE_TRANSFORM_OUTPUT_LENGTH = dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
 class TemplateTransformNode(Node[TemplateTransformNodeData]):
     node_type = NodeType.TEMPLATE_TRANSFORM
 
+    _node_data: TemplateTransformNodeData
+
     @classmethod
     def get_default_config(cls, filters: Mapping[str, object] | None = None) -> Mapping[str, object]:
         """
@@ -33,14 +35,14 @@ class TemplateTransformNode(Node[TemplateTransformNodeData]):
     def _run(self) -> NodeRunResult:
         # Get variables
         variables: dict[str, Any] = {}
-        for variable_selector in self.node_data.variables:
+        for variable_selector in self._node_data.variables:
             variable_name = variable_selector.variable
             value = self.graph_runtime_state.variable_pool.get(variable_selector.value_selector)
             variables[variable_name] = value.to_object() if value else None
         # Run code
         try:
             result = CodeExecutor.execute_workflow_code_template(
-                language=CodeLanguage.JINJA2, code=self.node_data.template, inputs=variables
+                language=CodeLanguage.JINJA2, code=self._node_data.template, inputs=variables
             )
         except CodeExecutionError as e:
             return NodeRunResult(inputs=variables, status=WorkflowNodeExecutionStatus.FAILED, error=str(e))

api/core/workflow/nodes/tool/entities.py

@@ -3,7 +3,7 @@ from typing import Any, Literal, Union
 from pydantic import BaseModel, field_validator
 from pydantic_core.core_schema import ValidationInfo
 
-from core.tools.entities.tool_entities import ToolProviderType
+from core.tools.entities.tool_entities import ToolAuthType, ToolProviderType
 from core.workflow.nodes.base.entities import BaseNodeData
 
 
@@ -16,6 +16,7 @@ class ToolEntity(BaseModel):
     tool_configurations: dict[str, Any]
     credential_id: str | None = None
     plugin_unique_identifier: str | None = None  # redundancy
+    auth_type: ToolAuthType = ToolAuthType.WORKSPACE  # OAuth authentication level
 
     @field_validator("tool_configurations", mode="before")
     @classmethod

api/core/workflow/nodes/tool/tool_node.py

@@ -12,6 +12,7 @@ from core.tools.entities.tool_entities import ToolInvokeMessage, ToolParameter
 from core.tools.errors import ToolInvokeError
 from core.tools.tool_engine import ToolEngine
 from core.tools.utils.message_transformer import ToolFileMessageTransformer
+from core.tools.workflow_as_tool.tool import WorkflowTool
 from core.variables.segments import ArrayAnySegment, ArrayFileSegment
 from core.variables.variables import ArrayAnyVariable
 from core.workflow.enums import (
@@ -46,6 +47,8 @@ class ToolNode(Node[ToolNodeData]):
 
     node_type = NodeType.TOOL
 
+    _node_data: ToolNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
@@ -56,11 +59,13 @@ class ToolNode(Node[ToolNodeData]):
         """
         from core.plugin.impl.exc import PluginDaemonClientSideError, PluginInvokeError
 
+        node_data = self._node_data
+
         # fetch tool icon
         tool_info = {
-            "provider_type": self.node_data.provider_type.value,
-            "provider_id": self.node_data.provider_id,
-            "plugin_unique_identifier": self.node_data.plugin_unique_identifier,
+            "provider_type": node_data.provider_type.value,
+            "provider_id": node_data.provider_id,
+            "plugin_unique_identifier": node_data.plugin_unique_identifier,
         }
 
         # get tool runtime
@@ -72,10 +77,10 @@ class ToolNode(Node[ToolNodeData]):
             # But for backward compatibility with historical data
             # this version field judgment is still preserved here.
             variable_pool: VariablePool | None = None
-            if self.node_data.version != "1" or self.node_data.tool_node_version is not None:
+            if node_data.version != "1" or node_data.tool_node_version is not None:
                 variable_pool = self.graph_runtime_state.variable_pool
             tool_runtime = ToolManager.get_workflow_tool_runtime(
-                self.tenant_id, self.app_id, self._node_id, self.node_data, self.invoke_from, variable_pool
+                self.tenant_id, self.app_id, self._node_id, self._node_data, self.invoke_from, variable_pool
             )
         except ToolNodeError as e:
             yield StreamCompletedEvent(
@@ -94,12 +99,12 @@ class ToolNode(Node[ToolNodeData]):
         parameters = self._generate_parameters(
             tool_parameters=tool_parameters,
             variable_pool=self.graph_runtime_state.variable_pool,
-            node_data=self.node_data,
+            node_data=self._node_data,
         )
         parameters_for_log = self._generate_parameters(
             tool_parameters=tool_parameters,
             variable_pool=self.graph_runtime_state.variable_pool,
-            node_data=self.node_data,
+            node_data=self._node_data,
             for_log=True,
         )
         # get conversation id
@@ -144,7 +149,7 @@ class ToolNode(Node[ToolNodeData]):
                     status=WorkflowNodeExecutionStatus.FAILED,
                     inputs=parameters_for_log,
                     metadata={WorkflowNodeExecutionMetadataKey.TOOL_INFO: tool_info},
-                    error=f"Failed to invoke tool {self.node_data.provider_name}: {str(e)}",
+                    error=f"Failed to invoke tool {node_data.provider_name}: {str(e)}",
                     error_type=type(e).__name__,
                 )
             )
@@ -154,7 +159,7 @@ class ToolNode(Node[ToolNodeData]):
                     status=WorkflowNodeExecutionStatus.FAILED,
                     inputs=parameters_for_log,
                     metadata={WorkflowNodeExecutionMetadataKey.TOOL_INFO: tool_info},
-                    error=e.to_user_friendly_error(plugin_name=self.node_data.provider_name),
+                    error=e.to_user_friendly_error(plugin_name=node_data.provider_name),
                     error_type=type(e).__name__,
                 )
             )
@@ -429,7 +434,7 @@ class ToolNode(Node[ToolNodeData]):
         metadata: dict[WorkflowNodeExecutionMetadataKey, Any] = {
             WorkflowNodeExecutionMetadataKey.TOOL_INFO: tool_info,
         }
-        if isinstance(usage.total_tokens, int) and usage.total_tokens > 0:
+        if usage.total_tokens > 0:
             metadata[WorkflowNodeExecutionMetadataKey.TOTAL_TOKENS] = usage.total_tokens
             metadata[WorkflowNodeExecutionMetadataKey.TOTAL_PRICE] = usage.total_price
             metadata[WorkflowNodeExecutionMetadataKey.CURRENCY] = usage.currency
@@ -448,17 +453,8 @@ class ToolNode(Node[ToolNodeData]):
 
     @staticmethod
     def _extract_tool_usage(tool_runtime: Tool) -> LLMUsage:
-        # Avoid importing WorkflowTool at module import time; rely on duck typing
-        # Some runtimes expose `latest_usage`; mocks may synthesize arbitrary attributes.
-        latest = getattr(tool_runtime, "latest_usage", None)
-        # Normalize into a concrete LLMUsage. MagicMock returns truthy attribute objects
-        # for any name, so we must type-check here.
-        if isinstance(latest, LLMUsage):
-            return latest
-        if isinstance(latest, dict):
-            # Allow dict payloads from external runtimes
-            return LLMUsage.model_validate(latest)
-        # Fallback to empty usage when attribute is missing or not a valid payload
+        if isinstance(tool_runtime, WorkflowTool):
+            return tool_runtime.latest_usage
         return LLMUsage.empty_usage()
 
     @classmethod
@@ -499,4 +495,4 @@ class ToolNode(Node[ToolNodeData]):
 
     @property
     def retry(self) -> bool:
-        return self.node_data.retry_config.retry_enabled
+        return self._node_data.retry_config.retry_enabled

api/core/workflow/nodes/trigger_plugin/trigger_event_node.py

@@ -43,9 +43,9 @@ class TriggerEventNode(Node[TriggerEventNodeData]):
         # Get trigger data passed when workflow was triggered
         metadata = {
             WorkflowNodeExecutionMetadataKey.TRIGGER_INFO: {
-                "provider_id": self.node_data.provider_id,
-                "event_name": self.node_data.event_name,
-                "plugin_unique_identifier": self.node_data.plugin_unique_identifier,
+                "provider_id": self._node_data.provider_id,
+                "event_name": self._node_data.event_name,
+                "plugin_unique_identifier": self._node_data.plugin_unique_identifier,
             },
         }
         node_inputs = dict(self.graph_runtime_state.variable_pool.user_inputs)

api/core/workflow/nodes/trigger_webhook/node.py

@@ -84,7 +84,7 @@ class TriggerWebhookNode(Node[WebhookData]):
         webhook_headers = webhook_data.get("headers", {})
         webhook_headers_lower = {k.lower(): v for k, v in webhook_headers.items()}
 
-        for header in self.node_data.headers:
+        for header in self._node_data.headers:
             header_name = header.name
             value = _get_normalized(webhook_headers, header_name)
             if value is None:
@@ -93,20 +93,20 @@ class TriggerWebhookNode(Node[WebhookData]):
             outputs[sanitized_name] = value
 
         # Extract configured query parameters
-        for param in self.node_data.params:
+        for param in self._node_data.params:
             param_name = param.name
             outputs[param_name] = webhook_data.get("query_params", {}).get(param_name)
 
         # Extract configured body parameters
-        for body_param in self.node_data.body:
+        for body_param in self._node_data.body:
             param_name = body_param.name
             param_type = body_param.type
 
-            if self.node_data.content_type == ContentType.TEXT:
+            if self._node_data.content_type == ContentType.TEXT:
                 # For text/plain, the entire body is a single string parameter
                 outputs[param_name] = str(webhook_data.get("body", {}).get("raw", ""))
                 continue
-            elif self.node_data.content_type == ContentType.BINARY:
+            elif self._node_data.content_type == ContentType.BINARY:
                 outputs[param_name] = webhook_data.get("body", {}).get("raw", b"")
                 continue
 

api/core/workflow/nodes/variable_aggregator/entities.py

@@ -23,11 +23,12 @@ class AdvancedSettings(BaseModel):
     groups: list[Group]
 
 
-class VariableAggregatorNodeData(BaseNodeData):
+class VariableAssignerNodeData(BaseNodeData):
     """
-    Variable Aggregator Node Data.
+    Variable Assigner Node Data.
     """
 
+    type: str = "variable-assigner"
     output_type: str
     variables: list[list[str]]
     advanced_settings: AdvancedSettings | None = None

api/core/workflow/nodes/variable_aggregator/variable_aggregator_node.py

@@ -4,12 +4,14 @@ from core.variables.segments import Segment
 from core.workflow.enums import NodeType, WorkflowNodeExecutionStatus
 from core.workflow.node_events import NodeRunResult
 from core.workflow.nodes.base.node import Node
-from core.workflow.nodes.variable_aggregator.entities import VariableAggregatorNodeData
+from core.workflow.nodes.variable_aggregator.entities import VariableAssignerNodeData
 
 
-class VariableAggregatorNode(Node[VariableAggregatorNodeData]):
+class VariableAggregatorNode(Node[VariableAssignerNodeData]):
     node_type = NodeType.VARIABLE_AGGREGATOR
 
+    _node_data: VariableAssignerNodeData
+
     @classmethod
     def version(cls) -> str:
         return "1"
@@ -19,8 +21,8 @@ class VariableAggregatorNode(Node[VariableAggregatorNodeData]):
         outputs: dict[str, Segment | Mapping[str, Segment]] = {}
         inputs = {}
 
-        if not self.node_data.advanced_settings or not self.node_data.advanced_settings.group_enabled:
-            for selector in self.node_data.variables:
+        if not self._node_data.advanced_settings or not self._node_data.advanced_settings.group_enabled:
+            for selector in self._node_data.variables:
                 variable = self.graph_runtime_state.variable_pool.get(selector)
                 if variable is not None:
                     outputs = {"output": variable}
@@ -28,7 +30,7 @@ class VariableAggregatorNode(Node[VariableAggregatorNodeData]):
                     inputs = {".".join(selector[1:]): variable.to_object()}
                     break
         else:
-            for group in self.node_data.advanced_settings.groups:
+            for group in self._node_data.advanced_settings.groups:
                 for selector in group.variables:
                     variable = self.graph_runtime_state.variable_pool.get(selector)
 

api/core/workflow/nodes/variable_assigner/v1/node.py

@@ -25,6 +25,8 @@ class VariableAssignerNode(Node[VariableAssignerData]):
     node_type = NodeType.VARIABLE_ASSIGNER
     _conv_var_updater_factory: _CONV_VAR_UPDATER_FACTORY
 
+    _node_data: VariableAssignerData
+
     def __init__(
         self,
         id: str,
@@ -69,21 +71,21 @@ class VariableAssignerNode(Node[VariableAssignerData]):
         return mapping
 
     def _run(self) -> NodeRunResult:
-        assigned_variable_selector = self.node_data.assigned_variable_selector
+        assigned_variable_selector = self._node_data.assigned_variable_selector
         # Should be String, Number, Object, ArrayString, ArrayNumber, ArrayObject
         original_variable = self.graph_runtime_state.variable_pool.get(assigned_variable_selector)
         if not isinstance(original_variable, Variable):
             raise VariableOperatorNodeError("assigned variable not found")
 
-        match self.node_data.write_mode:
+        match self._node_data.write_mode:
             case WriteMode.OVER_WRITE:
-                income_value = self.graph_runtime_state.variable_pool.get(self.node_data.input_variable_selector)
+                income_value = self.graph_runtime_state.variable_pool.get(self._node_data.input_variable_selector)
                 if not income_value:
                     raise VariableOperatorNodeError("input value not found")
                 updated_variable = original_variable.model_copy(update={"value": income_value.value})
 
             case WriteMode.APPEND:
-                income_value = self.graph_runtime_state.variable_pool.get(self.node_data.input_variable_selector)
+                income_value = self.graph_runtime_state.variable_pool.get(self._node_data.input_variable_selector)
                 if not income_value:
                     raise VariableOperatorNodeError("input value not found")
                 updated_value = original_variable.value + [income_value.value]

api/core/workflow/nodes/variable_assigner/v2/node.py

@@ -53,6 +53,8 @@ def _source_mapping_from_item(mapping: MutableMapping[str, Sequence[str]], node_
 class VariableAssignerNode(Node[VariableAssignerNodeData]):
     node_type = NodeType.VARIABLE_ASSIGNER
 
+    _node_data: VariableAssignerNodeData
+
     def blocks_variable_output(self, variable_selectors: set[tuple[str, ...]]) -> bool:
         """
         Check if this Variable Assigner node blocks the output of specific variables.
@@ -60,7 +62,7 @@ class VariableAssignerNode(Node[VariableAssignerNodeData]):
         Returns True if this node updates any of the requested conversation variables.
         """
         # Check each item in this Variable Assigner node
-        for item in self.node_data.items:
+        for item in self._node_data.items:
             # Convert the item's variable_selector to tuple for comparison
             item_selector_tuple = tuple(item.variable_selector)
 
@@ -95,13 +97,13 @@ class VariableAssignerNode(Node[VariableAssignerNodeData]):
         return var_mapping
 
     def _run(self) -> NodeRunResult:
-        inputs = self.node_data.model_dump()
+        inputs = self._node_data.model_dump()
         process_data: dict[str, Any] = {}
         # NOTE: This node has no outputs
         updated_variable_selectors: list[Sequence[str]] = []
 
         try:
-            for item in self.node_data.items:
+            for item in self._node_data.items:
                 variable = self.graph_runtime_state.variable_pool.get(item.variable_selector)
 
                 # ==================== Validation Part

api/extensions/ext_forward_refs.py

@@ -1,49 +0,0 @@
-import logging
-
-from dify_app import DifyApp
-
-
-def is_enabled() -> bool:
-    return True
-
-
-def init_app(app: DifyApp):
-    """Resolve Pydantic forward refs that would otherwise cause circular imports.
-
-    Rebuilds models in core.app.entities.app_invoke_entities with the real TraceQueueManager type.
-    Safe to run multiple times.
-    """
-    logger = logging.getLogger(__name__)
-    try:
-        from core.app.entities.app_invoke_entities import (
-            AdvancedChatAppGenerateEntity,
-            AgentChatAppGenerateEntity,
-            AppGenerateEntity,
-            ChatAppGenerateEntity,
-            CompletionAppGenerateEntity,
-            ConversationAppGenerateEntity,
-            EasyUIBasedAppGenerateEntity,
-            RagPipelineGenerateEntity,
-            WorkflowAppGenerateEntity,
-        )
-        from core.ops.ops_trace_manager import TraceQueueManager  # heavy import, do it at startup only
-
-        ns = {"TraceQueueManager": TraceQueueManager}
-        for Model in (
-            AppGenerateEntity,
-            EasyUIBasedAppGenerateEntity,
-            ConversationAppGenerateEntity,
-            ChatAppGenerateEntity,
-            CompletionAppGenerateEntity,
-            AgentChatAppGenerateEntity,
-            AdvancedChatAppGenerateEntity,
-            WorkflowAppGenerateEntity,
-            RagPipelineGenerateEntity,
-        ):
-            try:
-                Model.model_rebuild(_types_namespace=ns)
-            except Exception as e:
-                logger.debug("model_rebuild skipped for %s: %s", Model.__name__, e)
-    except Exception as e:
-        # Don't block app startup; just log at debug level.
-        logger.debug("ext_forward_refs init skipped: %s", e)

api/migrations/versions/2025_11_18_1400-a7b4e8f2c9d1_add_enduser_authentication_provider.py

@@ -0,0 +1,71 @@
+"""add enduser authentication provider
+
+Revision ID: a7b4e8f2c9d1
+Revises: fecff1c3da27
+Create Date: 2025-11-18 14:00:00.000000
+
+"""
+import models as models
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "a7b4e8f2c9d1"
+down_revision = "fecff1c3da27"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "tool_enduser_authentication_providers",
+        sa.Column(
+            "id",
+            models.types.StringUUID(),
+            nullable=False,
+        ),
+        sa.Column(
+            "name",
+            sa.String(length=256),
+            server_default="API KEY 1",
+            nullable=False,
+        ),
+        sa.Column("tenant_id", models.types.StringUUID(), nullable=False),
+        sa.Column("end_user_id", models.types.StringUUID(), nullable=False),
+        sa.Column("provider", sa.Text(), nullable=False),
+        sa.Column("encrypted_credentials", sa.Text(), default="", nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(),
+            server_default=sa.func.current_timestamp(),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(),
+            server_default=sa.func.current_timestamp(),
+            nullable=False,
+        ),
+        sa.Column(
+            "credential_type",
+            sa.String(length=32),
+            server_default="api-key",
+            nullable=False,
+        ),
+        sa.Column("expires_at", sa.BigInteger(), server_default=sa.text("-1"), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "end_user_id",
+            "provider",
+        ),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("tool_enduser_authentication_providers")
+    # ### end Alembic commands ###
+
+

api/migrations/versions/2025_12_15_1614-6bb0832495f0_alter_table_pipeline_recommended_.py

@@ -1,64 +0,0 @@
-"""Alter table pipeline_recommended_plugins add column type
-
-Revision ID: 6bb0832495f0
-Revises: 7bb281b7a422
-Create Date: 2025-12-15 16:14:38.482072
-
-"""
-from alembic import op
-import models as models
-import sqlalchemy as sa
-from sqlalchemy.dialects import postgresql
-
-# revision identifiers, used by Alembic.
-revision = '6bb0832495f0'
-down_revision = '7bb281b7a422'
-branch_labels = None
-depends_on = None
-
-
-def upgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('app_triggers', schema=None) as batch_op:
-        batch_op.alter_column('provider_name',
-               existing_type=sa.VARCHAR(length=255),
-               nullable=False,
-               existing_server_default=sa.text("''::character varying"))
-
-    with op.batch_alter_table('operation_logs', schema=None) as batch_op:
-        batch_op.alter_column('content',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=False)
-
-    with op.batch_alter_table('pipeline_recommended_plugins', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('type', sa.String(length=50), nullable=True))
-
-    with op.batch_alter_table('providers', schema=None) as batch_op:
-        batch_op.alter_column('quota_used',
-               existing_type=sa.BIGINT(),
-               nullable=False)
-
-    # ### end Alembic commands ###
-
-def downgrade():
-    # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('providers', schema=None) as batch_op:
-        batch_op.alter_column('quota_used',
-               existing_type=sa.BIGINT(),
-               nullable=True)
-
-    with op.batch_alter_table('pipeline_recommended_plugins', schema=None) as batch_op:
-        batch_op.drop_column('type')
-
-    with op.batch_alter_table('operation_logs', schema=None) as batch_op:
-        batch_op.alter_column('content',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=True)
-
-    with op.batch_alter_table('app_triggers', schema=None) as batch_op:
-        batch_op.alter_column('provider_name',
-               existing_type=sa.VARCHAR(length=255),
-               nullable=True,
-               existing_server_default=sa.text("''::character varying"))
-
-    # ### end Alembic commands ###

api/models/__init__.py

@@ -80,6 +80,7 @@ from .task import CeleryTask, CeleryTaskSet
 from .tools import (
     ApiToolProvider,
     BuiltinToolProvider,
+    EndUserAuthenticationProvider,
     ToolConversationVariables,
     ToolFile,
     ToolLabelBinding,
@@ -149,6 +150,7 @@ __all__ = [
     "DocumentSegment",
     "Embedding",
     "EndUser",
+    "EndUserAuthenticationProvider",
     "ExternalKnowledgeApis",
     "ExternalKnowledgeBindings",
     "IconType",

api/models/dataset.py

@@ -1458,7 +1458,6 @@ class PipelineRecommendedPlugin(TypeBase):
     )
     plugin_id: Mapped[str] = mapped_column(LongText, nullable=False)
     provider_name: Mapped[str] = mapped_column(LongText, nullable=False)
-    type: Mapped[str] = mapped_column(sa.String(50), nullable=True)
     position: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=0)
     active: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, default=True)
     created_at: Mapped[datetime] = mapped_column(

api/models/tools.py

@@ -9,9 +9,11 @@ from deprecated import deprecated
 from sqlalchemy import ForeignKey, String, func
 from sqlalchemy.orm import Mapped, mapped_column
 
+from core.plugin.entities.plugin_daemon import CredentialType
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_bundle import ApiToolBundle
 from core.tools.entities.tool_entities import ApiProviderSchemaType, WorkflowToolParameterConfiguration
+from libs.uuid_utils import uuidv7
 
 from .base import TypeBase
 from .engine import db
@@ -115,6 +117,59 @@ class BuiltinToolProvider(TypeBase):
         return cast(dict[str, Any], json.loads(self.encrypted_credentials))
 
 
+class EndUserAuthenticationProvider(TypeBase):
+    """
+    This table stores the authentication credentials for end users in tools.
+    Mimics the BuiltinToolProvider structure but for end users instead of tenants.
+    """
+
+    __tablename__ = "tool_enduser_authentication_providers"
+    __table_args__ = (
+        sa.UniqueConstraint("end_user_id", "provider"),
+    )
+
+    # id of the authentication provider
+    id: Mapped[str] = mapped_column(StringUUID, primary_key=True, default=lambda: str(uuidv7()), init=False)
+    # id of the tenant
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    # id of the end user
+    end_user_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    # name of the tool provider
+    provider: Mapped[str] = mapped_column(LongText, nullable=False)
+    name: Mapped[str] = mapped_column(
+        String(256),
+        nullable=False,
+        default="API KEY 1",
+    )
+    # encrypted credentials for the end user
+    encrypted_credentials: Mapped[str] = mapped_column(LongText, nullable=False, default="")
+    created_at: Mapped[datetime] = mapped_column(
+        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        sa.DateTime,
+        nullable=False,
+        server_default=func.current_timestamp(),
+        onupdate=func.current_timestamp(),
+        init=False,
+    )
+    # credential type, e.g., "api-key", "oauth2"
+    credential_type: Mapped[CredentialType] = mapped_column(
+        String(32), nullable=False, default=CredentialType.API_KEY
+    )
+    # Unix timestamp in seconds since epoch (1970-01-01 UTC); -1 indicates no expiration
+    expires_at: Mapped[int] = mapped_column(sa.BigInteger, nullable=False, default=-1)
+
+    @property
+    def credentials(self) -> dict[str, Any]:
+        if not self.encrypted_credentials:
+            return {}
+        try:
+            return cast(dict[str, Any], json.loads(self.encrypted_credentials))
+        except json.JSONDecodeError:
+            return {}
+
+
 class ApiToolProvider(TypeBase):
     """
     The table stores the api providers.

api/models/types.py

@@ -19,7 +19,7 @@ class StringUUID(TypeDecorator[uuid.UUID | str | None]):
     def process_bind_param(self, value: uuid.UUID | str | None, dialect: Dialect) -> str | None:
         if value is None:
             return value
-        elif dialect.name in ["postgresql", "mysql"]:
+        elif dialect.name == "postgresql":
             return str(value)
         else:
             if isinstance(value, uuid.UUID):

api/models/workflow.py

@@ -907,29 +907,19 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
     @property
     def extras(self) -> dict[str, Any]:
         from core.tools.tool_manager import ToolManager
-        from core.trigger.trigger_manager import TriggerManager
 
         extras: dict[str, Any] = {}
-        execution_metadata = self.execution_metadata_dict
-        if execution_metadata:
-            if self.node_type == NodeType.TOOL and "tool_info" in execution_metadata:
-                tool_info: dict[str, Any] = execution_metadata["tool_info"]
+        if self.execution_metadata_dict:
+            if self.node_type == NodeType.TOOL and "tool_info" in self.execution_metadata_dict:
+                tool_info: dict[str, Any] = self.execution_metadata_dict["tool_info"]
                 extras["icon"] = ToolManager.get_tool_icon(
                     tenant_id=self.tenant_id,
                     provider_type=tool_info["provider_type"],
                     provider_id=tool_info["provider_id"],
                 )
-            elif self.node_type == NodeType.DATASOURCE and "datasource_info" in execution_metadata:
-                datasource_info = execution_metadata["datasource_info"]
+            elif self.node_type == NodeType.DATASOURCE and "datasource_info" in self.execution_metadata_dict:
+                datasource_info = self.execution_metadata_dict["datasource_info"]
                 extras["icon"] = datasource_info.get("icon")
-            elif self.node_type == NodeType.TRIGGER_PLUGIN and "trigger_info" in execution_metadata:
-                trigger_info = execution_metadata["trigger_info"] or {}
-                provider_id = trigger_info.get("provider_id")
-                if provider_id:
-                    extras["icon"] = TriggerManager.get_trigger_plugin_icon(
-                        tenant_id=self.tenant_id,
-                        provider_id=provider_id,
-                    )
         return extras
 
     def _get_offload_by_type(self, type_: ExecutionOffLoadType) -> Optional["WorkflowNodeExecutionOffload"]:

api/pyproject.toml

@@ -111,7 +111,7 @@ package = false
 dev = [
     "coverage~=7.2.4",
     "dotenv-linter~=0.5.0",
-    "faker~=38.2.0",
+    "faker~=32.1.0",
     "lxml-stubs~=0.5.1",
     "ty~=0.0.1a19",
     "basedpyright~=1.31.0",

api/services/feedback_service.py

@@ -86,7 +86,7 @@ class FeedbackService:
         export_data = []
         for feedback, message, conversation, app, account in results:
             # Get the user query from the message
-            user_query = message.query or (message.inputs.get("query", "") if message.inputs else "")
+            user_query = message.query or message.inputs.get("query", "") if message.inputs else ""
 
             # Format the feedback data
             feedback_record = {

api/services/rag_pipeline/rag_pipeline.py

@@ -1248,14 +1248,12 @@ class RagPipelineService:
             session.commit()
         return workflow_node_execution_db_model
 
-    def get_recommended_plugins(self, type: str) -> dict:
+    def get_recommended_plugins(self) -> dict:
         # Query active recommended plugins
-        query = db.session.query(PipelineRecommendedPlugin).where(PipelineRecommendedPlugin.active == True)
-        if type and type != "all":
-            query = query.where(PipelineRecommendedPlugin.type == type)
-
         pipeline_recommended_plugins = (
-            query.order_by(PipelineRecommendedPlugin.position.asc())
+            db.session.query(PipelineRecommendedPlugin)
+            .where(PipelineRecommendedPlugin.active == True)
+            .order_by(PipelineRecommendedPlugin.position.asc())
             .all()
         )
 

api/services/tools/tools_transform_service.py

@@ -201,9 +201,7 @@ class ToolTransformService:
 
     @staticmethod
     def workflow_provider_to_user_provider(
-        provider_controller: WorkflowToolProviderController,
-        labels: list[str] | None = None,
-        workflow_app_id: str | None = None,
+        provider_controller: WorkflowToolProviderController, labels: list[str] | None = None
     ):
         """
         convert provider controller to user provider
@@ -223,7 +221,6 @@ class ToolTransformService:
             plugin_unique_identifier=None,
             tools=[],
             labels=labels or [],
-            workflow_app_id=workflow_app_id,
         )
 
     @staticmethod

api/services/tools/workflow_tools_manage_service.py

@@ -189,9 +189,6 @@ class WorkflowToolManageService:
             select(WorkflowToolProvider).where(WorkflowToolProvider.tenant_id == tenant_id)
         ).all()
 
-        # Create a mapping from provider_id to app_id
-        provider_id_to_app_id = {provider.id: provider.app_id for provider in db_tools}
-
         tools: list[WorkflowToolProviderController] = []
         for provider in db_tools:
             try:
@@ -205,11 +202,8 @@ class WorkflowToolManageService:
         result = []
 
         for tool in tools:
-            workflow_app_id = provider_id_to_app_id.get(tool.provider_id)
             user_tool_provider = ToolTransformService.workflow_provider_to_user_provider(
-                provider_controller=tool,
-                labels=labels.get(tool.provider_id, []),
-                workflow_app_id=workflow_app_id,
+                provider_controller=tool, labels=labels.get(tool.provider_id, [])
             )
             ToolTransformService.repack_provider(tenant_id=tenant_id, provider=user_tool_provider)
             user_tool_provider.tools = [

api/tests/unit_tests/core/rag/embedding/__init__.py

@@ -1 +0,0 @@
-"""Unit tests for core.rag.embedding module."""

api/tests/unit_tests/core/tools/entities/test_api_entities.py

@@ -1,100 +0,0 @@
-"""
-Unit tests for ToolProviderApiEntity workflow_app_id field.
-
-This test suite covers:
-- ToolProviderApiEntity workflow_app_id field creation and default value
-- ToolProviderApiEntity.to_dict() method behavior with workflow_app_id
-"""
-
-from core.tools.entities.api_entities import ToolProviderApiEntity
-from core.tools.entities.common_entities import I18nObject
-from core.tools.entities.tool_entities import ToolProviderType
-
-
-class TestToolProviderApiEntityWorkflowAppId:
-    """Test suite for ToolProviderApiEntity workflow_app_id field."""
-
-    def test_workflow_app_id_field_default_none(self):
-        """Test that workflow_app_id defaults to None when not provided."""
-        entity = ToolProviderApiEntity(
-            id="test_id",
-            author="test_author",
-            name="test_name",
-            description=I18nObject(en_US="Test description"),
-            icon="test_icon",
-            label=I18nObject(en_US="Test label"),
-            type=ToolProviderType.WORKFLOW,
-        )
-
-        assert entity.workflow_app_id is None
-
-    def test_to_dict_includes_workflow_app_id_when_workflow_type_and_has_value(self):
-        """Test that to_dict() includes workflow_app_id when type is WORKFLOW and value is set."""
-        workflow_app_id = "app_123"
-        entity = ToolProviderApiEntity(
-            id="test_id",
-            author="test_author",
-            name="test_name",
-            description=I18nObject(en_US="Test description"),
-            icon="test_icon",
-            label=I18nObject(en_US="Test label"),
-            type=ToolProviderType.WORKFLOW,
-            workflow_app_id=workflow_app_id,
-        )
-
-        result = entity.to_dict()
-
-        assert "workflow_app_id" in result
-        assert result["workflow_app_id"] == workflow_app_id
-
-    def test_to_dict_excludes_workflow_app_id_when_workflow_type_and_none(self):
-        """Test that to_dict() excludes workflow_app_id when type is WORKFLOW but value is None."""
-        entity = ToolProviderApiEntity(
-            id="test_id",
-            author="test_author",
-            name="test_name",
-            description=I18nObject(en_US="Test description"),
-            icon="test_icon",
-            label=I18nObject(en_US="Test label"),
-            type=ToolProviderType.WORKFLOW,
-            workflow_app_id=None,
-        )
-
-        result = entity.to_dict()
-
-        assert "workflow_app_id" not in result
-
-    def test_to_dict_excludes_workflow_app_id_when_not_workflow_type(self):
-        """Test that to_dict() excludes workflow_app_id when type is not WORKFLOW."""
-        workflow_app_id = "app_123"
-        entity = ToolProviderApiEntity(
-            id="test_id",
-            author="test_author",
-            name="test_name",
-            description=I18nObject(en_US="Test description"),
-            icon="test_icon",
-            label=I18nObject(en_US="Test label"),
-            type=ToolProviderType.BUILT_IN,
-            workflow_app_id=workflow_app_id,
-        )
-
-        result = entity.to_dict()
-
-        assert "workflow_app_id" not in result
-
-    def test_to_dict_includes_workflow_app_id_for_workflow_type_with_empty_string(self):
-        """Test that to_dict() excludes workflow_app_id when value is empty string (falsy)."""
-        entity = ToolProviderApiEntity(
-            id="test_id",
-            author="test_author",
-            name="test_name",
-            description=I18nObject(en_US="Test description"),
-            icon="test_icon",
-            label=I18nObject(en_US="Test label"),
-            type=ToolProviderType.WORKFLOW,
-            workflow_app_id="",
-        )
-
-        result = entity.to_dict()
-
-        assert "workflow_app_id" not in result

api/tests/unit_tests/core/workflow/graph/test_graph_validation.py

@@ -29,7 +29,7 @@ class _TestNode(Node[_TestNodeData]):
 
     @classmethod
     def version(cls) -> str:
-        return "1"
+        return "test"
 
     def __init__(
         self,

api/tests/unit_tests/core/workflow/graph_engine/test_graph_engine.py

@@ -744,7 +744,7 @@ def test_graph_run_emits_partial_success_when_node_failure_recovered():
     )
 
     llm_node = graph.nodes["llm"]
-    base_node_data = llm_node.node_data
+    base_node_data = llm_node.get_base_node_data()
     base_node_data.error_strategy = ErrorStrategy.DEFAULT_VALUE
     base_node_data.default_value = [DefaultValue(key="text", value="fallback response", type=DefaultValueType.STRING)]
 

api/tests/unit_tests/core/workflow/graph_engine/test_mock_nodes.py

@@ -92,7 +92,7 @@ class MockLLMNode(MockNodeMixin, LLMNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock LLM node."""
@@ -189,7 +189,7 @@ class MockAgentNode(MockNodeMixin, AgentNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock agent node."""
@@ -241,7 +241,7 @@ class MockToolNode(MockNodeMixin, ToolNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock tool node."""
@@ -294,7 +294,7 @@ class MockKnowledgeRetrievalNode(MockNodeMixin, KnowledgeRetrievalNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock knowledge retrieval node."""
@@ -351,7 +351,7 @@ class MockHttpRequestNode(MockNodeMixin, HttpRequestNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock HTTP request node."""
@@ -404,7 +404,7 @@ class MockQuestionClassifierNode(MockNodeMixin, QuestionClassifierNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock question classifier node."""
@@ -452,7 +452,7 @@ class MockParameterExtractorNode(MockNodeMixin, ParameterExtractorNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock parameter extractor node."""
@@ -502,7 +502,7 @@ class MockDocumentExtractorNode(MockNodeMixin, DocumentExtractorNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> Generator:
         """Execute mock document extractor node."""
@@ -557,7 +557,7 @@ class MockIterationNode(MockNodeMixin, IterationNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _create_graph_engine(self, index: int, item: Any):
         """Create a graph engine with MockNodeFactory instead of DifyNodeFactory."""
@@ -632,7 +632,7 @@ class MockLoopNode(MockNodeMixin, LoopNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _create_graph_engine(self, start_at, root_node_id: str):
         """Create a graph engine with MockNodeFactory instead of DifyNodeFactory."""
@@ -694,7 +694,7 @@ class MockTemplateTransformNode(MockNodeMixin, TemplateTransformNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> NodeRunResult:
         """Execute mock template transform node."""
@@ -780,7 +780,7 @@ class MockCodeNode(MockNodeMixin, CodeNode):
     @classmethod
     def version(cls) -> str:
         """Return the version of this mock node."""
-        return "1"
+        return "mock-1"
 
     def _run(self) -> NodeRunResult:
         """Execute mock code node."""

api/tests/unit_tests/core/workflow/nodes/base/test_base_node.py

@@ -33,10 +33,6 @@ def test_ensure_subclasses_of_base_node_has_node_type_and_version_method_defined
     type_version_set: set[tuple[NodeType, str]] = set()
 
     for cls in classes:
-        # Only validate production node classes; skip test-defined subclasses and external helpers
-        module_name = getattr(cls, "__module__", "")
-        if not module_name.startswith("core."):
-            continue
         # Validate that 'version' is directly defined in the class (not inherited) by checking the class's __dict__
         assert "version" in cls.__dict__, f"class {cls} should have version method defined (NOT INHERITED.)"
         node_type = cls.node_type

api/tests/unit_tests/core/workflow/nodes/base/test_get_node_type_classes_mapping.py

@@ -1,84 +0,0 @@
-import types
-from collections.abc import Mapping
-
-from core.workflow.enums import NodeType
-from core.workflow.nodes.base.entities import BaseNodeData
-from core.workflow.nodes.base.node import Node
-
-# Import concrete nodes we will assert on (numeric version path)
-from core.workflow.nodes.variable_assigner.v1.node import (
-    VariableAssignerNode as VariableAssignerV1,
-)
-from core.workflow.nodes.variable_assigner.v2.node import (
-    VariableAssignerNode as VariableAssignerV2,
-)
-
-
-def test_variable_assigner_latest_prefers_highest_numeric_version():
-    # Act
-    mapping: Mapping[NodeType, Mapping[str, type[Node]]] = Node.get_node_type_classes_mapping()
-
-    # Assert basic presence
-    assert NodeType.VARIABLE_ASSIGNER in mapping
-    va_versions = mapping[NodeType.VARIABLE_ASSIGNER]
-
-    # Both concrete versions must be present
-    assert va_versions.get("1") is VariableAssignerV1
-    assert va_versions.get("2") is VariableAssignerV2
-
-    # And latest should point to numerically-highest version ("2")
-    assert va_versions.get("latest") is VariableAssignerV2
-
-
-def test_latest_prefers_highest_numeric_version():
-    # Arrange: define two ephemeral subclasses with numeric versions under a NodeType
-    # that has no concrete implementations in production to avoid interference.
-    class _Version1(Node[BaseNodeData]):  # type: ignore[misc]
-        node_type = NodeType.LEGACY_VARIABLE_AGGREGATOR
-
-        def init_node_data(self, data):
-            pass
-
-        def _run(self):
-            raise NotImplementedError
-
-        @classmethod
-        def version(cls) -> str:
-            return "1"
-
-        def _get_error_strategy(self):
-            return None
-
-        def _get_retry_config(self):
-            return types.SimpleNamespace()  # not used
-
-        def _get_title(self) -> str:
-            return "version1"
-
-        def _get_description(self):
-            return None
-
-        def _get_default_value_dict(self):
-            return {}
-
-        def get_base_node_data(self):
-            return types.SimpleNamespace(title="version1")
-
-    class _Version2(_Version1):  # type: ignore[misc]
-        @classmethod
-        def version(cls) -> str:
-            return "2"
-
-        def _get_title(self) -> str:
-            return "version2"
-
-    # Act: build a fresh mapping (it should now see our ephemeral subclasses)
-    mapping: Mapping[NodeType, Mapping[str, type[Node]]] = Node.get_node_type_classes_mapping()
-
-    # Assert: both numeric versions exist for this NodeType; 'latest' points to the higher numeric version
-    assert NodeType.LEGACY_VARIABLE_AGGREGATOR in mapping
-    legacy_versions = mapping[NodeType.LEGACY_VARIABLE_AGGREGATOR]
-
-    assert legacy_versions.get("1") is _Version1
-    assert legacy_versions.get("2") is _Version2
-    assert legacy_versions.get("latest") is _Version2

api/tests/unit_tests/core/workflow/nodes/code/code_node_spec.py

@@ -471,8 +471,8 @@ class TestCodeNodeInitialization:
 
         assert node._get_description() is None
 
-    def test_node_data_property(self):
-        """Test node_data property returns node data."""
+    def test_get_base_node_data(self):
+        """Test get_base_node_data returns node data."""
         node = CodeNode.__new__(CodeNode)
         node._node_data = CodeNodeData(
             title="Base Test",
@@ -482,7 +482,7 @@ class TestCodeNodeInitialization:
             outputs={},
         )
 
-        result = node.node_data
+        result = node.get_base_node_data()
 
         assert result == node._node_data
         assert result.title == "Base Test"

api/tests/unit_tests/core/workflow/nodes/iteration/iteration_node_spec.py

@@ -240,8 +240,8 @@ class TestIterationNodeInitialization:
 
         assert node._get_description() == "This is a description"
 
-    def test_node_data_property(self):
-        """Test node_data property returns node data."""
+    def test_get_base_node_data(self):
+        """Test get_base_node_data returns node data."""
         node = IterationNode.__new__(IterationNode)
         node._node_data = IterationNodeData(
             title="Base Test",
@@ -249,7 +249,7 @@ class TestIterationNodeInitialization:
             output_selector=["y"],
         )
 
-        result = node.node_data
+        result = node.get_base_node_data()
 
         assert result == node._node_data