doc: add debugging steps in api/README.md

fix: not from markpalceplace show auto update
chore: remove volumns field in docker-compose-template.yaml for worker_beat service
2025-07-17 15:35:37 +08:00 · 2025-07-17 10:35:05 +08:00 · 2025-07-11 15:36:08 +08:00 · 2025-07-11 15:34:17 +08:00 · 2025-07-11 15:32:34 +08:00 · 2025-07-11 15:22:51 +08:00
1094 changed files with 45180 additions and 9963 deletions
@@ -47,15 +47,17 @@ jobs:
      - name: Run Unit tests
        run: |
          uv run --project api bash dev/pytest/pytest_unit_tests.sh
+
+      - name: Coverage Summary
+        run: |
+          set -x
          # Extract coverage percentage and create a summary
          TOTAL_COVERAGE=$(python -c 'import json; print(json.load(open("coverage.json"))["totals"]["percent_covered_display"])')

          # Create a detailed coverage summary
          echo "### Test Coverage Summary :test_tube:" >> $GITHUB_STEP_SUMMARY
          echo "Total Coverage: ${TOTAL_COVERAGE}%" >> $GITHUB_STEP_SUMMARY
-          echo "\`\`\`" >> $GITHUB_STEP_SUMMARY
-          uv run --project api coverage report >> $GITHUB_STEP_SUMMARY
-          echo "\`\`\`" >> $GITHUB_STEP_SUMMARY
+          uv run --project api coverage report --format=markdown >> $GITHUB_STEP_SUMMARY

      - name: Run dify config tests
        run: uv run --project api dev/pytest/pytest_config_tests.py
@@ -83,9 +85,15 @@ jobs:
          compose-file: |
            docker/docker-compose.middleware.yaml
          services: |
+            db
+            redis
            sandbox
            ssrf_proxy

+      - name: setup test config
+        run: |
+          cp api/tests/integration_tests/.env.example api/tests/integration_tests/.env
+
      - name: Run Workflow
        run: uv run --project api bash dev/pytest/pytest_workflow.sh

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

@@ -84,6 +84,12 @@ jobs:
            elasticsearch
            oceanbase

+      - name: setup test config
+        run: |
+          echo $(pwd)
+          ls -lah .
+          cp api/tests/integration_tests/.env.example api/tests/integration_tests/.env
+
      - name: Check VDB Ready (TiDB)
        run: uv run --project api python api/tests/integration_tests/vdb/tidb_vector/check_tiflash_ready.py

@@ -214,3 +214,4 @@ mise.toml

 # AI Assistant
 .roo/
+api/.env.backup
@@ -230,6 +230,10 @@ Deploy Dify to AWS with [CDK](https://aws.amazon.com/cdk/)

 Quickly deploy Dify to Alibaba cloud with [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88) 

+#### Using Alibaba Cloud Data Management
+
+One-Click deploy Dify to Alibaba Cloud with [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/) 
+

 ## Contributing

@@ -211,6 +211,11 @@ docker compose up -d

 #### استخدام Alibaba Cloud للنشر
 [بسرعة نشر Dify إلى سحابة علي بابا مع عش الحوسبة السحابية علي بابا](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)
+ 
+#### استخدام Alibaba Cloud Data Management للنشر
+
+انشر Dify على علي بابا كلاود بنقرة واحدة باستخدام [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## المساهمة

@@ -229,6 +229,10 @@ GitHub-এ ডিফাইকে স্টার দিয়ে রাখুন

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management ব্যবহার করে ডিপ্লয়
+
+ [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## Contributing

@@ -225,6 +225,10 @@ docker compose up -d

 使用 [阿里云计算巢](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88) 将 Dify 一键部署到 阿里云

+#### 使用 阿里云数据管理DMS 部署
+
+使用 [阿里云数据管理DMS](https://help.aliyun.com/zh/dms/dify-in-invitational-preview) 将 Dify 一键部署到 阿里云
+

 ## Star History

@@ -225,6 +225,10 @@ Bereitstellung von Dify auf AWS mit [CDK](https://aws.amazon.com/cdk/)

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+Ein-Klick-Bereitstellung von Dify in der Alibaba Cloud mit [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## Contributing

@@ -225,6 +225,11 @@ Despliegue Dify en AWS usando [CDK](https://aws.amazon.com/cdk/)

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+Despliega Dify en Alibaba Cloud con un solo clic con [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+
+
 ## Contribuir

 Para aquellos que deseen contribuir con código, consulten nuestra [Guía de contribución](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md). 
@@ -223,6 +223,10 @@ Déployez Dify sur AWS en utilisant [CDK](https://aws.amazon.com/cdk/)

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+Déployez Dify en un clic sur Alibaba Cloud avec [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## Contribuer

@@ -155,7 +155,7 @@ DifyはオープンソースのLLMアプリケーション開発プラットフ
 [こちら](https://dify.ai)のDify Cloudサービスを利用して、セットアップ不要で試すことができます。サンドボックスプランには、200回のGPT-4呼び出しが無料で含まれています。

 - **Dify Community Editionのセルフホスティング</br>**
-この[スタートガイド](#quick-start)を使用して、ローカル環境でDifyを簡単に実行できます。
+この[スタートガイド](#クイックスタート)を使用して、ローカル環境でDifyを簡単に実行できます。
 詳しくは[ドキュメント](https://docs.dify.ai)をご覧ください。

 - **企業/組織向けのDify</br>**
@@ -223,6 +223,9 @@ docker compose up -d
 #### Alibaba Cloud
 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+[Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/) を利用して、DifyをAlibaba Cloudへワンクリックでデプロイできます
+

 ## 貢献

@@ -223,6 +223,10 @@ wa'logh nIqHom neH ghun deployment toy'wI' [CDK](https://aws.amazon.com/cdk/) lo

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+[Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## Contributing

@@ -217,6 +217,10 @@ Dify를 Kubernetes에 배포하고 프리미엄 스케일링 설정을 구성했

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+[Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)를 통해 원클릭으로 Dify를 Alibaba Cloud에 배포할 수 있습니다
+

 ## 기여

@@ -222,6 +222,10 @@ Implante o Dify na AWS usando [CDK](https://aws.amazon.com/cdk/)

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+Implante o Dify na Alibaba Cloud com um clique usando o [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## Contribuindo

@@ -223,6 +223,10 @@ Uvedite Dify v AWS z uporabo [CDK](https://aws.amazon.com/cdk/)

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+Z enim klikom namestite Dify na Alibaba Cloud z [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## Prispevam

@@ -216,6 +216,10 @@ Dify'ı bulut platformuna tek tıklamayla dağıtın [terraform](https://www.ter

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+[Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/) kullanarak Dify'ı tek tıkla Alibaba Cloud'a dağıtın
+

 ## Katkıda Bulunma

@@ -228,6 +228,10 @@ Dify 的所有功能都提供相應的 API，因此您可以輕鬆地將 Dify

 [阿里云](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### 使用 阿里雲數據管理DMS 進行部署
+
+透過 [阿里雲數據管理DMS](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)，一鍵將 Dify 部署至阿里雲
+

 ## 貢獻

@@ -219,6 +219,10 @@ Triển khai Dify trên AWS bằng [CDK](https://aws.amazon.com/cdk/)

 [Alibaba Cloud Computing Nest](https://computenest.console.aliyun.com/service/instance/create/default?type=user&ServiceName=Dify%E7%A4%BE%E5%8C%BA%E7%89%88)

+#### Alibaba Cloud Data Management
+
+Triển khai Dify lên Alibaba Cloud chỉ với một cú nhấp chuột bằng [Alibaba Cloud Data Management](https://www.alibabacloud.com/help/en/dms/dify-in-invitational-preview/)
+

 ## Đóng góp

@@ -451,6 +451,16 @@ APP_MAX_ACTIVE_REQUESTS=0
 # Celery beat configuration
 CELERY_BEAT_SCHEDULER_TIME=1

+# Celery schedule tasks configuration
+ENABLE_CLEAN_EMBEDDING_CACHE_TASK=false
+ENABLE_CLEAN_UNUSED_DATASETS_TASK=false
+ENABLE_CREATE_TIDB_SERVERLESS_TASK=false
+ENABLE_UPDATE_TIDB_SERVERLESS_STATUS_TASK=false
+ENABLE_CLEAN_MESSAGES=false
+ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK=false
+ENABLE_DATASETS_QUEUE_MONITOR=false
+ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK=true
+
 # Position configuration
 POSITION_TOOL_PINS=
 POSITION_TOOL_INCLUDES=
@@ -1,6 +1,4 @@
-exclude = [
-    "migrations/*",
-]
+exclude = ["migrations/*"]
 line-length = 120

 [format]
@@ -9,14 +7,14 @@ quote-style = "double"
 [lint]
 preview = false
 select = [
-    "B", # flake8-bugbear rules
-    "C4", # flake8-comprehensions
-    "E", # pycodestyle E rules
-    "F", # pyflakes rules
-    "FURB", # refurb rules
-    "I", # isort rules
-    "N", # pep8-naming
-    "PT", # flake8-pytest-style rules
+    "B",       # flake8-bugbear rules
+    "C4",      # flake8-comprehensions
+    "E",       # pycodestyle E rules
+    "F",       # pyflakes rules
+    "FURB",    # refurb rules
+    "I",       # isort rules
+    "N",       # pep8-naming
+    "PT",      # flake8-pytest-style rules
    "PLC0208", # iteration-over-set
    "PLC0414", # useless-import-alias
    "PLE0604", # invalid-all-object
@@ -24,19 +22,19 @@ select = [
    "PLR0402", # manual-from-import
    "PLR1711", # useless-return
    "PLR1714", # repeated-equality-comparison
-    "RUF013", # implicit-optional
-    "RUF019", # unnecessary-key-check
-    "RUF100", # unused-noqa
-    "RUF101", # redirected-noqa
-    "RUF200", # invalid-pyproject-toml
-    "RUF022", # unsorted-dunder-all
-    "S506", # unsafe-yaml-load
-    "SIM", # flake8-simplify rules
-    "TRY400", # error-instead-of-exception
-    "TRY401", # verbose-log-message
-    "UP", # pyupgrade rules
-    "W191", # tab-indentation
-    "W605", # invalid-escape-sequence
+    "RUF013",  # implicit-optional
+    "RUF019",  # unnecessary-key-check
+    "RUF100",  # unused-noqa
+    "RUF101",  # redirected-noqa
+    "RUF200",  # invalid-pyproject-toml
+    "RUF022",  # unsorted-dunder-all
+    "S506",    # unsafe-yaml-load
+    "SIM",     # flake8-simplify rules
+    "TRY400",  # error-instead-of-exception
+    "TRY401",  # verbose-log-message
+    "UP",      # pyupgrade rules
+    "W191",    # tab-indentation
+    "W605",    # invalid-escape-sequence
    # security related linting rules
    # RCE proctection (sort of)
    "S102", # exec-builtin, disallow use of `exec`
@@ -47,36 +45,37 @@ select = [
 ]

 ignore = [
-    "E402", # module-import-not-at-top-of-file
-    "E711", # none-comparison
-    "E712", # true-false-comparison
-    "E721", # type-comparison
-    "E722", # bare-except
-    "F821", # undefined-name
-    "F841", # unused-variable
+    "E402",    # module-import-not-at-top-of-file
+    "E711",    # none-comparison
+    "E712",    # true-false-comparison
+    "E721",    # type-comparison
+    "E722",    # bare-except
+    "F821",    # undefined-name
+    "F841",    # unused-variable
    "FURB113", # repeated-append
    "FURB152", # math-constant
-    "UP007", # non-pep604-annotation
-    "UP032", # f-string
-    "UP045", # non-pep604-annotation-optional
-    "B005", # strip-with-multi-characters
-    "B006", # mutable-argument-default
-    "B007", # unused-loop-control-variable
-    "B026", # star-arg-unpacking-after-keyword-arg
-    "B903", # class-as-data-structure
-    "B904", # raise-without-from-inside-except
-    "B905", # zip-without-explicit-strict
-    "N806", # non-lowercase-variable-in-function
-    "N815", # mixed-case-variable-in-class-scope
-    "PT011", # pytest-raises-too-broad
-    "SIM102", # collapsible-if
-    "SIM103", # needless-bool
-    "SIM105", # suppressible-exception
-    "SIM107", # return-in-try-except-finally
-    "SIM108", # if-else-block-instead-of-if-exp
-    "SIM113", # enumerate-for-loop
-    "SIM117", # multiple-with-statements
-    "SIM210", # if-expr-with-true-false
+    "UP007",   # non-pep604-annotation
+    "UP032",   # f-string
+    "UP045",   # non-pep604-annotation-optional
+    "B005",    # strip-with-multi-characters
+    "B006",    # mutable-argument-default
+    "B007",    # unused-loop-control-variable
+    "B026",    # star-arg-unpacking-after-keyword-arg
+    "B903",    # class-as-data-structure
+    "B904",    # raise-without-from-inside-except
+    "B905",    # zip-without-explicit-strict
+    "N806",    # non-lowercase-variable-in-function
+    "N815",    # mixed-case-variable-in-class-scope
+    "PT011",   # pytest-raises-too-broad
+    "SIM102",  # collapsible-if
+    "SIM103",  # needless-bool
+    "SIM105",  # suppressible-exception
+    "SIM107",  # return-in-try-except-finally
+    "SIM108",  # if-else-block-instead-of-if-exp
+    "SIM113",  # enumerate-for-loop
+    "SIM117",  # multiple-with-statements
+    "SIM210",  # if-expr-with-true-false
+    "UP038",   # deprecated and not recommended by Ruff, https://docs.astral.sh/ruff/rules/non-pep604-isinstance/
 ]

 [lint.per-file-ignores]
@@ -74,7 +74,12 @@
 10. If you need to handle and debug the async tasks (e.g. dataset importing and documents indexing), please start the worker service.

   ```bash
-   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion
+   uv run celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion,plugin
+   ```
+
+   Addition, if you want to debug the celery scheduled tasks, you can use the following command in another terminal:
+   ```bash
+   uv run celery -A app.celery beat 
   ```

 ## Testing
@@ -1,8 +1,11 @@
 import logging
+from pathlib import Path
 from typing import Any

 from pydantic.fields import FieldInfo
-from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict
+from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, TomlConfigSettingsSource
+
+from libs.file_utils import search_file_upwards

 from .deploy import DeploymentConfig
 from .enterprise import EnterpriseFeatureConfig
@@ -99,4 +102,12 @@ class DifyConfig(
            RemoteSettingsSourceFactory(settings_cls),
            dotenv_settings,
            file_secret_settings,
+            TomlConfigSettingsSource(
+                settings_cls=settings_cls,
+                toml_file=search_file_upwards(
+                    base_dir_path=Path(__file__).parent,
+                    target_file_name="pyproject.toml",
+                    max_search_parent_depth=2,
+                ),
+            ),
        )
@@ -779,6 +779,41 @@ class CeleryBeatConfig(BaseSettings):
    )


+class CeleryScheduleTasksConfig(BaseSettings):
+    ENABLE_CLEAN_EMBEDDING_CACHE_TASK: bool = Field(
+        description="Enable clean embedding cache task",
+        default=False,
+    )
+    ENABLE_CLEAN_UNUSED_DATASETS_TASK: bool = Field(
+        description="Enable clean unused datasets task",
+        default=False,
+    )
+    ENABLE_CREATE_TIDB_SERVERLESS_TASK: bool = Field(
+        description="Enable create tidb service job task",
+        default=False,
+    )
+    ENABLE_UPDATE_TIDB_SERVERLESS_STATUS_TASK: bool = Field(
+        description="Enable update tidb service job status task",
+        default=False,
+    )
+    ENABLE_CLEAN_MESSAGES: bool = Field(
+        description="Enable clean messages task",
+        default=False,
+    )
+    ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK: bool = Field(
+        description="Enable mail clean document notify task",
+        default=False,
+    )
+    ENABLE_DATASETS_QUEUE_MONITOR: bool = Field(
+        description="Enable queue monitor task",
+        default=False,
+    )
+    ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK: bool = Field(
+        description="Enable check upgradable plugin task",
+        default=True,
+    )
+
+
 class PositionConfig(BaseSettings):
    POSITION_PROVIDER_PINS: str = Field(
        description="Comma-separated list of pinned model providers",
@@ -907,5 +942,6 @@ class FeatureConfig(
    # hosted services config
    HostedServiceConfig,
    CeleryBeatConfig,
+    CeleryScheduleTasksConfig,
 ):
    pass
@@ -223,6 +223,10 @@ class CeleryConfig(DatabaseConfig):
        default=None,
    )

+    CELERY_SENTINEL_PASSWORD: Optional[str] = Field(
+        description="Password of the Redis Sentinel master.",
+        default=None,
+    )
    CELERY_SENTINEL_SOCKET_TIMEOUT: Optional[PositiveFloat] = Field(
        description="Timeout for Redis Sentinel socket operations in seconds.",
        default=0.1,
@@ -1,17 +1,13 @@
 from pydantic import Field
-from pydantic_settings import BaseSettings
+
+from configs.packaging.pyproject import PyProjectConfig, PyProjectTomlConfig


-class PackagingInfo(BaseSettings):
+class PackagingInfo(PyProjectTomlConfig):
    """
    Packaging build information
    """

-    CURRENT_VERSION: str = Field(
-        description="Dify version",
-        default="1.4.3",
-    )
-
    COMMIT_SHA: str = Field(
        description="SHA-1 checksum of the git commit used to build the app",
        default="",
@@ -0,0 +1,17 @@
+from pydantic import BaseModel, Field
+from pydantic_settings import BaseSettings
+
+
+class PyProjectConfig(BaseModel):
+    version: str = Field(description="Dify version", default="")
+
+
+class PyProjectTomlConfig(BaseSettings):
+    """
+    configs in api/pyproject.toml
+    """
+
+    project: PyProjectConfig = Field(
+        description="configs in the project section of pyproject.toml",
+        default=PyProjectConfig(),
+    )
@@ -56,6 +56,7 @@ from .app import (
    conversation,
    conversation_variables,
    generator,
+    mcp_server,
    message,
    model_config,
    ops_trace,
@@ -63,6 +64,7 @@ from .app import (
    statistic,
    workflow,
    workflow_app_log,
+    workflow_draft_variable,
    workflow_run,
    workflow_statistic,
 )
@@ -90,23 +90,11 @@ class ChatMessageTextApi(Resource):

            message_id = args.get("message_id", None)
            text = args.get("text", None)
-            if (
-                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
-                and app_model.workflow
-                and app_model.workflow.features_dict
-            ):
-                text_to_speech = app_model.workflow.features_dict.get("text_to_speech")
-                if text_to_speech is None:
-                    raise ValueError("TTS is not enabled")
-                voice = args.get("voice") or text_to_speech.get("voice")
-            else:
-                try:
-                    if app_model.app_model_config is None:
-                        raise ValueError("AppModelConfig not found")
-                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
-                except Exception:
-                    voice = None
-            response = AudioService.transcript_tts(app_model=app_model, text=text, message_id=message_id, voice=voice)
+            voice = args.get("voice", None)
+
+            response = AudioService.transcript_tts(
+                app_model=app_model, text=text, voice=voice, message_id=message_id, is_draft=True
+            )
            return response
        except services.errors.app_model_config.AppModelConfigBrokenError:
            logging.exception("App model config broken.")
@@ -0,0 +1,102 @@
+import json
+from enum import StrEnum
+
+from flask_login import current_user
+from flask_restful import Resource, marshal_with, reqparse
+from werkzeug.exceptions import NotFound
+
+from controllers.console import api
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, setup_required
+from extensions.ext_database import db
+from fields.app_fields import app_server_fields
+from libs.login import login_required
+from models.model import AppMCPServer
+
+
+class AppMCPServerStatus(StrEnum):
+    ACTIVE = "active"
+    INACTIVE = "inactive"
+
+
+class AppMCPServerController(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(app_server_fields)
+    def get(self, app_model):
+        server = db.session.query(AppMCPServer).filter(AppMCPServer.app_id == app_model.id).first()
+        return server
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(app_server_fields)
+    def post(self, app_model):
+        # The role of the current user in the ta table must be editor, admin, or owner
+        if not current_user.is_editor:
+            raise NotFound()
+        parser = reqparse.RequestParser()
+        parser.add_argument("description", type=str, required=True, location="json")
+        parser.add_argument("parameters", type=dict, required=True, location="json")
+        args = parser.parse_args()
+        server = AppMCPServer(
+            name=app_model.name,
+            description=args["description"],
+            parameters=json.dumps(args["parameters"], ensure_ascii=False),
+            status=AppMCPServerStatus.ACTIVE,
+            app_id=app_model.id,
+            tenant_id=current_user.current_tenant_id,
+            server_code=AppMCPServer.generate_server_code(16),
+        )
+        db.session.add(server)
+        db.session.commit()
+        return server
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model
+    @marshal_with(app_server_fields)
+    def put(self, app_model):
+        if not current_user.is_editor:
+            raise NotFound()
+        parser = reqparse.RequestParser()
+        parser.add_argument("id", type=str, required=True, location="json")
+        parser.add_argument("description", type=str, required=True, location="json")
+        parser.add_argument("parameters", type=dict, required=True, location="json")
+        parser.add_argument("status", type=str, required=False, location="json")
+        args = parser.parse_args()
+        server = db.session.query(AppMCPServer).filter(AppMCPServer.id == args["id"]).first()
+        if not server:
+            raise NotFound()
+        server.description = args["description"]
+        server.parameters = json.dumps(args["parameters"], ensure_ascii=False)
+        if args["status"]:
+            if args["status"] not in [status.value for status in AppMCPServerStatus]:
+                raise ValueError("Invalid status")
+            server.status = args["status"]
+        db.session.commit()
+        return server
+
+
+class AppMCPServerRefreshController(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @marshal_with(app_server_fields)
+    def get(self, server_id):
+        if not current_user.is_editor:
+            raise NotFound()
+        server = db.session.query(AppMCPServer).filter(AppMCPServer.id == server_id).first()
+        if not server:
+            raise NotFound()
+        server.server_code = AppMCPServer.generate_server_code(16)
+        db.session.commit()
+        return server
+
+
+api.add_resource(AppMCPServerController, "/apps/<uuid:app_id>/server")
+api.add_resource(AppMCPServerRefreshController, "/apps/<uuid:server_id>/server/refresh")
@@ -1,5 +1,6 @@
 import json
 import logging
+from collections.abc import Sequence
 from typing import cast

 from flask import abort, request
@@ -18,10 +19,12 @@ from controllers.console.app.error import (
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
+from core.app.app_config.features.file_upload.manager import FileUploadConfigManager
 from core.app.apps.base_app_queue_manager import AppQueueManager
 from core.app.entities.app_invoke_entities import InvokeFrom
+from core.file.models import File
 from extensions.ext_database import db
-from factories import variable_factory
+from factories import file_factory, variable_factory
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from fields.workflow_run_fields import workflow_run_node_execution_fields
 from libs import helper
@@ -30,6 +33,7 @@ from libs.login import current_user, login_required
 from models import App
 from models.account import Account
 from models.model import AppMode
+from models.workflow import Workflow
 from services.app_generate_service import AppGenerateService
 from services.errors.app import WorkflowHashNotEqualError
 from services.errors.llm import InvokeRateLimitError
@@ -38,6 +42,24 @@ from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseE
 logger = logging.getLogger(__name__)


+# 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.
+def _parse_file(workflow: Workflow, files: list[dict] | None = None) -> Sequence[File]:
+    files = files or []
+
+    file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
+    file_objs: Sequence[File] = []
+    if file_extra_config is None:
+        return file_objs
+    file_objs = file_factory.build_from_mappings(
+        mappings=files,
+        tenant_id=workflow.tenant_id,
+        config=file_extra_config,
+    )
+    return file_objs
+
+
 class DraftWorkflowApi(Resource):
    @setup_required
    @login_required
@@ -402,15 +424,30 @@ class DraftWorkflowNodeRunApi(Resource):

        parser = reqparse.RequestParser()
        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("query", type=str, required=False, location="json", default="")
+        parser.add_argument("files", type=list, location="json", default=[])
        args = parser.parse_args()

-        inputs = args.get("inputs")
-        if inputs == None:
+        user_inputs = args.get("inputs")
+        if user_inputs is None:
            raise ValueError("missing inputs")

+        workflow_srv = WorkflowService()
+        # fetch draft workflow by app_model
+        draft_workflow = workflow_srv.get_draft_workflow(app_model=app_model)
+        if not draft_workflow:
+            raise ValueError("Workflow not initialized")
+        files = _parse_file(draft_workflow, args.get("files"))
        workflow_service = WorkflowService()
+
        workflow_node_execution = workflow_service.run_draft_workflow_node(
-            app_model=app_model, node_id=node_id, user_inputs=inputs, account=current_user
+            app_model=app_model,
+            draft_workflow=draft_workflow,
+            node_id=node_id,
+            user_inputs=user_inputs,
+            account=current_user,
+            query=args.get("query", ""),
+            files=files,
        )

        return workflow_node_execution
@@ -731,6 +768,27 @@ class WorkflowByIdApi(Resource):
        return None, 204


+class DraftWorkflowNodeLastRunApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    @marshal_with(workflow_run_node_execution_fields)
+    def get(self, app_model: App, node_id: str):
+        srv = WorkflowService()
+        workflow = srv.get_draft_workflow(app_model)
+        if not workflow:
+            raise NotFound("Workflow not found")
+        node_exec = srv.get_node_last_run(
+            app_model=app_model,
+            workflow=workflow,
+            node_id=node_id,
+        )
+        if node_exec is None:
+            raise NotFound("last run not found")
+        return node_exec
+
+
 api.add_resource(
    DraftWorkflowApi,
    "/apps/<uuid:app_id>/workflows/draft",
@@ -795,3 +853,7 @@ api.add_resource(
    WorkflowByIdApi,
    "/apps/<uuid:app_id>/workflows/<string:workflow_id>",
 )
+api.add_resource(
+    DraftWorkflowNodeLastRunApi,
+    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run",
+)
@@ -0,0 +1,421 @@
+import logging
+from typing import Any, NoReturn
+
+from flask import Response
+from flask_restful import Resource, fields, inputs, marshal, marshal_with, reqparse
+from sqlalchemy.orm import Session
+from werkzeug.exceptions import Forbidden
+
+from controllers.console import api
+from controllers.console.app.error import (
+    DraftWorkflowNotExist,
+)
+from controllers.console.app.wraps import get_app_model
+from controllers.console.wraps import account_initialization_required, setup_required
+from controllers.web.error import InvalidArgumentError, NotFoundError
+from core.variables.segment_group import SegmentGroup
+from core.variables.segments import ArrayFileSegment, FileSegment, Segment
+from core.variables.types import SegmentType
+from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
+from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
+from libs.login import current_user, login_required
+from models import App, AppMode, db
+from models.workflow import WorkflowDraftVariable
+from services.workflow_draft_variable_service import WorkflowDraftVariableList, WorkflowDraftVariableService
+from services.workflow_service import WorkflowService
+
+logger = logging.getLogger(__name__)
+
+
+def _convert_values_to_json_serializable_object(value: Segment) -> Any:
+    if isinstance(value, FileSegment):
+        return value.value.model_dump()
+    elif isinstance(value, ArrayFileSegment):
+        return [i.model_dump() for i in value.value]
+    elif isinstance(value, SegmentGroup):
+        return [_convert_values_to_json_serializable_object(i) for i in value.value]
+    else:
+        return value.value
+
+
+def _serialize_var_value(variable: WorkflowDraftVariable) -> Any:
+    value = variable.get_value()
+    # create a copy of the value to avoid affecting the model cache.
+    value = value.model_copy(deep=True)
+    # Refresh the url signature before returning it to client.
+    if isinstance(value, FileSegment):
+        file = value.value
+        file.remote_url = file.generate_url()
+    elif isinstance(value, ArrayFileSegment):
+        files = value.value
+        for file in files:
+            file.remote_url = file.generate_url()
+    return _convert_values_to_json_serializable_object(value)
+
+
+def _create_pagination_parser():
+    parser = reqparse.RequestParser()
+    parser.add_argument(
+        "page",
+        type=inputs.int_range(1, 100_000),
+        required=False,
+        default=1,
+        location="args",
+        help="the page of data requested",
+    )
+    parser.add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args")
+    return parser
+
+
+_WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS = {
+    "id": fields.String,
+    "type": fields.String(attribute=lambda model: model.get_variable_type()),
+    "name": fields.String,
+    "description": fields.String,
+    "selector": fields.List(fields.String, attribute=lambda model: model.get_selector()),
+    "value_type": fields.String,
+    "edited": fields.Boolean(attribute=lambda model: model.edited),
+    "visible": fields.Boolean,
+}
+
+_WORKFLOW_DRAFT_VARIABLE_FIELDS = dict(
+    _WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS,
+    value=fields.Raw(attribute=_serialize_var_value),
+)
+
+_WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS = {
+    "id": fields.String,
+    "type": fields.String(attribute=lambda _: "env"),
+    "name": fields.String,
+    "description": fields.String,
+    "selector": fields.List(fields.String, attribute=lambda model: model.get_selector()),
+    "value_type": fields.String,
+    "edited": fields.Boolean(attribute=lambda model: model.edited),
+    "visible": fields.Boolean,
+}
+
+_WORKFLOW_DRAFT_ENV_VARIABLE_LIST_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_ENV_VARIABLE_FIELDS)),
+}
+
+
+def _get_items(var_list: WorkflowDraftVariableList) -> list[WorkflowDraftVariable]:
+    return var_list.variables
+
+
+_WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_VARIABLE_WITHOUT_VALUE_FIELDS), attribute=_get_items),
+    "total": fields.Raw(),
+}
+
+_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS = {
+    "items": fields.List(fields.Nested(_WORKFLOW_DRAFT_VARIABLE_FIELDS), attribute=_get_items),
+}
+
+
+def _api_prerequisite(f):
+    """Common prerequisites for all draft workflow variable APIs.
+
+    It ensures the following conditions are satisfied:
+
+    - Dify has been property setup.
+    - The request user has logged in and initialized.
+    - The requested app is a workflow or a chat flow.
+    - The request user has the edit permission for the app.
+    """
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
+    def wrapper(*args, **kwargs):
+        if not current_user.is_editor:
+            raise Forbidden()
+        return f(*args, **kwargs)
+
+    return wrapper
+
+
+class WorkflowVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_WITHOUT_VALUE_FIELDS)
+    def get(self, app_model: App):
+        """
+        Get draft workflow
+        """
+        parser = _create_pagination_parser()
+        args = parser.parse_args()
+
+        # fetch draft workflow by app_model
+        workflow_service = WorkflowService()
+        workflow_exist = workflow_service.is_workflow_exist(app_model=app_model)
+        if not workflow_exist:
+            raise DraftWorkflowNotExist()
+
+        # fetch draft workflow by app_model
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(
+                session=session,
+            )
+        workflow_vars = draft_var_srv.list_variables_without_values(
+            app_id=app_model.id,
+            page=args.page,
+            limit=args.limit,
+        )
+
+        return workflow_vars
+
+    @_api_prerequisite
+    def delete(self, app_model: App):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        draft_var_srv.delete_workflow_variables(app_model.id)
+        db.session.commit()
+        return Response("", 204)
+
+
+def validate_node_id(node_id: str) -> NoReturn | None:
+    if node_id in [
+        CONVERSATION_VARIABLE_NODE_ID,
+        SYSTEM_VARIABLE_NODE_ID,
+    ]:
+        # NOTE(QuantumGhost): While we store the system and conversation variables as node variables
+        # with specific `node_id` in database, we still want to make the API separated. By disallowing
+        # accessing system and conversation variables in `WorkflowDraftNodeVariableListApi`,
+        # we mitigate the risk that user of the API depending on the implementation detail of the API.
+        #
+        # ref: [Hyrum's Law](https://www.hyrumslaw.com/)
+
+        raise InvalidArgumentError(
+            f"invalid node_id, please use correspond api for conversation and system variables, node_id={node_id}",
+        )
+    return None
+
+
+class NodeVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
+    def get(self, app_model: App, node_id: str):
+        validate_node_id(node_id)
+        with Session(bind=db.engine, expire_on_commit=False) as session:
+            draft_var_srv = WorkflowDraftVariableService(
+                session=session,
+            )
+            node_vars = draft_var_srv.list_node_variables(app_model.id, node_id)
+
+        return node_vars
+
+    @_api_prerequisite
+    def delete(self, app_model: App, node_id: str):
+        validate_node_id(node_id)
+        srv = WorkflowDraftVariableService(db.session())
+        srv.delete_node_variables(app_model.id, node_id)
+        db.session.commit()
+        return Response("", 204)
+
+
+class VariableApi(Resource):
+    _PATCH_NAME_FIELD = "name"
+    _PATCH_VALUE_FIELD = "value"
+
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
+    def get(self, app_model: App, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        return variable
+
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_FIELDS)
+    def patch(self, app_model: App, variable_id: str):
+        # Request payload for file types:
+        #
+        # Local File:
+        #
+        #     {
+        #         "type": "image",
+        #         "transfer_method": "local_file",
+        #         "url": "",
+        #         "upload_file_id": "daded54f-72c7-4f8e-9d18-9b0abdd9f190"
+        #     }
+        #
+        # Remote File:
+        #
+        #
+        #     {
+        #         "type": "image",
+        #         "transfer_method": "remote_url",
+        #         "url": "http://127.0.0.1:5001/files/1602650a-4fe4-423c-85a2-af76c083e3c4/file-preview?timestamp=1750041099&nonce=...&sign=...=",
+        #         "upload_file_id": "1602650a-4fe4-423c-85a2-af76c083e3c4"
+        #     }
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(self._PATCH_NAME_FIELD, type=str, required=False, nullable=True, location="json")
+        # Parse 'value' field as-is to maintain its original data structure
+        parser.add_argument(self._PATCH_VALUE_FIELD, type=lambda x: x, required=False, nullable=True, location="json")
+
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        args = parser.parse_args(strict=True)
+
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+        new_name = args.get(self._PATCH_NAME_FIELD, None)
+        raw_value = args.get(self._PATCH_VALUE_FIELD, None)
+        if new_name is None and raw_value is None:
+            return variable
+
+        new_value = None
+        if raw_value is not None:
+            if variable.value_type == SegmentType.FILE:
+                if not isinstance(raw_value, dict):
+                    raise InvalidArgumentError(description=f"expected dict for file, got {type(raw_value)}")
+                raw_value = build_from_mapping(mapping=raw_value, tenant_id=app_model.tenant_id)
+            elif variable.value_type == SegmentType.ARRAY_FILE:
+                if not isinstance(raw_value, list):
+                    raise InvalidArgumentError(description=f"expected list for files, got {type(raw_value)}")
+                if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
+                    raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
+                raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
+            new_value = build_segment_with_type(variable.value_type, raw_value)
+        draft_var_srv.update_variable(variable, name=new_name, value=new_value)
+        db.session.commit()
+        return variable
+
+    @_api_prerequisite
+    def delete(self, app_model: App, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        draft_var_srv.delete_variable(variable)
+        db.session.commit()
+        return Response("", 204)
+
+
+class VariableResetApi(Resource):
+    @_api_prerequisite
+    def put(self, app_model: App, variable_id: str):
+        draft_var_srv = WorkflowDraftVariableService(
+            session=db.session(),
+        )
+
+        workflow_srv = WorkflowService()
+        draft_workflow = workflow_srv.get_draft_workflow(app_model)
+        if draft_workflow is None:
+            raise NotFoundError(
+                f"Draft workflow not found, app_id={app_model.id}",
+            )
+        variable = draft_var_srv.get_variable(variable_id=variable_id)
+        if variable is None:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+        if variable.app_id != app_model.id:
+            raise NotFoundError(description=f"variable not found, id={variable_id}")
+
+        resetted = draft_var_srv.reset_variable(draft_workflow, variable)
+        db.session.commit()
+        if resetted is None:
+            return Response("", 204)
+        else:
+            return marshal(resetted, _WORKFLOW_DRAFT_VARIABLE_FIELDS)
+
+
+def _get_variable_list(app_model: App, node_id) -> WorkflowDraftVariableList:
+    with Session(bind=db.engine, expire_on_commit=False) as session:
+        draft_var_srv = WorkflowDraftVariableService(
+            session=session,
+        )
+        if node_id == CONVERSATION_VARIABLE_NODE_ID:
+            draft_vars = draft_var_srv.list_conversation_variables(app_model.id)
+        elif node_id == SYSTEM_VARIABLE_NODE_ID:
+            draft_vars = draft_var_srv.list_system_variables(app_model.id)
+        else:
+            draft_vars = draft_var_srv.list_node_variables(app_id=app_model.id, node_id=node_id)
+    return draft_vars
+
+
+class ConversationVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
+    def get(self, app_model: App):
+        # NOTE(QuantumGhost): Prefill conversation variables into the draft variables table
+        # so their IDs can be returned to the caller.
+        workflow_srv = WorkflowService()
+        draft_workflow = workflow_srv.get_draft_workflow(app_model)
+        if draft_workflow is None:
+            raise NotFoundError(description=f"draft workflow not found, id={app_model.id}")
+        draft_var_srv = WorkflowDraftVariableService(db.session())
+        draft_var_srv.prefill_conversation_variable_default_values(draft_workflow)
+        db.session.commit()
+        return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
+
+
+class SystemVariableCollectionApi(Resource):
+    @_api_prerequisite
+    @marshal_with(_WORKFLOW_DRAFT_VARIABLE_LIST_FIELDS)
+    def get(self, app_model: App):
+        return _get_variable_list(app_model, SYSTEM_VARIABLE_NODE_ID)
+
+
+class EnvironmentVariableCollectionApi(Resource):
+    @_api_prerequisite
+    def get(self, app_model: App):
+        """
+        Get draft workflow
+        """
+        # fetch draft workflow by app_model
+        workflow_service = WorkflowService()
+        workflow = workflow_service.get_draft_workflow(app_model=app_model)
+        if workflow is None:
+            raise DraftWorkflowNotExist()
+
+        env_vars = workflow.environment_variables
+        env_vars_list = []
+        for v in env_vars:
+            env_vars_list.append(
+                {
+                    "id": v.id,
+                    "type": "env",
+                    "name": v.name,
+                    "description": v.description,
+                    "selector": v.selector,
+                    "value_type": v.value_type.value,
+                    "value": v.value,
+                    # Do not track edited for env vars.
+                    "edited": False,
+                    "visible": True,
+                    "editable": True,
+                }
+            )
+
+        return {"items": env_vars_list}
+
+
+api.add_resource(
+    WorkflowVariableCollectionApi,
+    "/apps/<uuid:app_id>/workflows/draft/variables",
+)
+api.add_resource(NodeVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
+api.add_resource(VariableApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
+api.add_resource(VariableResetApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
+
+api.add_resource(ConversationVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/conversation-variables")
+api.add_resource(SystemVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/system-variables")
+api.add_resource(EnvironmentVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/environment-variables")
@@ -8,6 +8,15 @@ from libs.login import current_user
 from models import App, AppMode


+def _load_app_model(app_id: str) -> Optional[App]:
+    app_model = (
+        db.session.query(App)
+        .filter(App.id == app_id, App.tenant_id == current_user.current_tenant_id, App.status == "normal")
+        .first()
+    )
+    return app_model
+
+
 def get_app_model(view: Optional[Callable] = None, *, mode: Union[AppMode, list[AppMode], None] = None):
    def decorator(view_func):
        @wraps(view_func)
@@ -20,11 +29,7 @@ def get_app_model(view: Optional[Callable] = None, *, mode: Union[AppMode, list[

            del kwargs["app_id"]

-            app_model = (
-                db.session.query(App)
-                .filter(App.id == app_id, App.tenant_id == current_user.current_tenant_id, App.status == "normal")
-                .first()
-            )
+            app_model = _load_app_model(app_id)

            if not app_model:
                raise AppNotFoundError()
@@ -41,7 +41,7 @@ class OAuthDataSource(Resource):
            if not internal_secret:
                return ({"error": "Internal secret is not set"},)
            oauth_provider.save_internal_access_token(internal_secret)
-            return {"data": ""}
+            return {"data": "internal"}
        else:
            auth_url = oauth_provider.get_authorization_url()
            return {"data": auth_url}, 200
@@ -5,7 +5,7 @@ from typing import cast

 from flask import request
 from flask_login import current_user
-from flask_restful import Resource, fields, marshal, marshal_with, reqparse
+from flask_restful import Resource, marshal, marshal_with, reqparse
 from sqlalchemy import asc, desc, select
 from werkzeug.exceptions import Forbidden, NotFound

@@ -239,12 +239,10 @@ class DatasetDocumentListApi(Resource):

        return response

-    documents_and_batch_fields = {"documents": fields.List(fields.Nested(document_fields)), "batch": fields.String}
-
    @setup_required
    @login_required
    @account_initialization_required
-    @marshal_with(documents_and_batch_fields)
+    @marshal_with(dataset_and_document_fields)
    @cloud_edition_billing_resource_check("vector_space")
    @cloud_edition_billing_rate_limit_check("knowledge")
    def post(self, dataset_id):
@@ -290,6 +288,8 @@ class DatasetDocumentListApi(Resource):

        try:
            documents, batch = DocumentService.save_document_with_dataset_id(dataset, knowledge_config, current_user)
+            dataset = DatasetService.get_dataset(dataset_id)
+
        except ProviderTokenNotInitError as ex:
            raise ProviderNotInitializeError(ex.description)
        except QuotaExceededError:
@@ -297,7 +297,7 @@ class DatasetDocumentListApi(Resource):
        except ModelCurrentlyNotSupportError:
            raise ProviderModelCurrentlyNotSupportError()

-        return {"documents": documents, "batch": batch}
+        return {"dataset": dataset, "documents": documents, "batch": batch}

    @setup_required
    @login_required
@@ -18,7 +18,6 @@ from controllers.console.app.error import (
 from controllers.console.explore.wraps import InstalledAppResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeError
-from models.model import AppMode
 from services.audio_service import AudioService
 from services.errors.audio import (
    AudioTooLargeServiceError,
@@ -79,19 +78,9 @@ class ChatTextApi(InstalledAppResource):

            message_id = args.get("message_id", None)
            text = args.get("text", None)
-            if (
-                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
-                and app_model.workflow
-                and app_model.workflow.features_dict
-            ):
-                text_to_speech = app_model.workflow.features_dict.get("text_to_speech")
-                voice = args.get("voice") or text_to_speech.get("voice")
-            else:
-                try:
-                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
-                except Exception:
-                    voice = None
-            response = AudioService.transcript_tts(app_model=app_model, message_id=message_id, voice=voice, text=text)
+            voice = args.get("voice", None)
+
+            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
            return response
        except services.errors.app_model_config.AppModelConfigBrokenError:
            logging.exception("App model config broken.")
@@ -18,7 +18,7 @@ class VersionApi(Resource):
        check_update_url = dify_config.CHECK_UPDATE_URL

        result = {
-            "version": dify_config.CURRENT_VERSION,
+            "version": dify_config.project.version,
            "release_date": "",
            "release_notes": "",
            "can_auto_update": False,
@@ -85,6 +85,7 @@ class MemberInviteEmailApi(Resource):
        return {
            "result": "success",
            "invitation_results": invitation_results,
+            "tenant_id": str(current_user.current_tenant.id),
        }, 201


@@ -110,7 +111,7 @@ class MemberCancelInviteApi(Resource):
            except Exception as e:
                raise ValueError(str(e))

-        return {"result": "success"}, 204
+        return {"result": "success", "tenant_id": str(current_user.current_tenant.id)}, 200


 class MemberUpdateRoleApi(Resource):
@@ -12,7 +12,9 @@ from controllers.console.wraps import account_initialization_required, setup_req
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from libs.login import login_required
-from models.account import TenantPluginPermission
+from models.account import TenantPluginAutoUpgradeStrategy, TenantPluginPermission
+from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
+from services.plugin.plugin_parameter_service import PluginParameterService
 from services.plugin.plugin_permission_service import PluginPermissionService
 from services.plugin.plugin_service import PluginService

@@ -497,6 +499,150 @@ class PluginFetchPermissionApi(Resource):
        )


+class PluginFetchDynamicSelectOptionsApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        # check if the user is admin or owner
+        if not current_user.is_admin_or_owner:
+            raise Forbidden()
+
+        tenant_id = current_user.current_tenant_id
+        user_id = current_user.id
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("plugin_id", type=str, required=True, location="args")
+        parser.add_argument("provider", type=str, required=True, location="args")
+        parser.add_argument("action", type=str, required=True, location="args")
+        parser.add_argument("parameter", type=str, required=True, location="args")
+        parser.add_argument("provider_type", type=str, required=True, location="args")
+        args = parser.parse_args()
+
+        try:
+            options = PluginParameterService.get_dynamic_select_options(
+                tenant_id,
+                user_id,
+                args["plugin_id"],
+                args["provider"],
+                args["action"],
+                args["parameter"],
+                args["provider_type"],
+            )
+        except PluginDaemonClientSideError as e:
+            raise ValueError(e)
+
+        return jsonable_encoder({"options": options})
+
+
+class PluginChangePreferencesApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        user = current_user
+        if not user.is_admin_or_owner:
+            raise Forbidden()
+
+        req = reqparse.RequestParser()
+        req.add_argument("permission", type=dict, required=True, location="json")
+        req.add_argument("auto_upgrade", type=dict, required=True, location="json")
+        args = req.parse_args()
+
+        tenant_id = user.current_tenant_id
+
+        permission = args["permission"]
+
+        install_permission = TenantPluginPermission.InstallPermission(permission.get("install_permission", "everyone"))
+        debug_permission = TenantPluginPermission.DebugPermission(permission.get("debug_permission", "everyone"))
+
+        auto_upgrade = args["auto_upgrade"]
+
+        strategy_setting = TenantPluginAutoUpgradeStrategy.StrategySetting(
+            auto_upgrade.get("strategy_setting", "fix_only")
+        )
+        upgrade_time_of_day = auto_upgrade.get("upgrade_time_of_day", 0)
+        upgrade_mode = TenantPluginAutoUpgradeStrategy.UpgradeMode(auto_upgrade.get("upgrade_mode", "exclude"))
+        exclude_plugins = auto_upgrade.get("exclude_plugins", [])
+        include_plugins = auto_upgrade.get("include_plugins", [])
+
+        # set permission
+        set_permission_result = PluginPermissionService.change_permission(
+            tenant_id,
+            install_permission,
+            debug_permission,
+        )
+        if not set_permission_result:
+            return jsonable_encoder({"success": False, "message": "Failed to set permission"})
+
+        # set auto upgrade strategy
+        set_auto_upgrade_strategy_result = PluginAutoUpgradeService.change_strategy(
+            tenant_id,
+            strategy_setting,
+            upgrade_time_of_day,
+            upgrade_mode,
+            exclude_plugins,
+            include_plugins,
+        )
+        if not set_auto_upgrade_strategy_result:
+            return jsonable_encoder({"success": False, "message": "Failed to set auto upgrade strategy"})
+
+        return jsonable_encoder({"success": True})
+
+
+class PluginFetchPreferencesApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        tenant_id = current_user.current_tenant_id
+
+        permission = PluginPermissionService.get_permission(tenant_id)
+        permission_dict = {
+            "install_permission": TenantPluginPermission.InstallPermission.EVERYONE,
+            "debug_permission": TenantPluginPermission.DebugPermission.EVERYONE,
+        }
+
+        if permission:
+            permission_dict["install_permission"] = permission.install_permission
+            permission_dict["debug_permission"] = permission.debug_permission
+
+        auto_upgrade = PluginAutoUpgradeService.get_strategy(tenant_id)
+        auto_upgrade_dict = {
+            "strategy_setting": TenantPluginAutoUpgradeStrategy.StrategySetting.DISABLED,
+            "upgrade_time_of_day": 0,
+            "upgrade_mode": TenantPluginAutoUpgradeStrategy.UpgradeMode.EXCLUDE,
+            "exclude_plugins": [],
+            "include_plugins": [],
+        }
+
+        if auto_upgrade:
+            auto_upgrade_dict = {
+                "strategy_setting": auto_upgrade.strategy_setting,
+                "upgrade_time_of_day": auto_upgrade.upgrade_time_of_day,
+                "upgrade_mode": auto_upgrade.upgrade_mode,
+                "exclude_plugins": auto_upgrade.exclude_plugins,
+                "include_plugins": auto_upgrade.include_plugins,
+            }
+
+        return jsonable_encoder({"permission": permission_dict, "auto_upgrade": auto_upgrade_dict})
+
+
+class PluginAutoUpgradeExcludePluginApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        # exclude one single plugin
+        tenant_id = current_user.current_tenant_id
+
+        req = reqparse.RequestParser()
+        req.add_argument("plugin_id", type=str, required=True, location="json")
+        args = req.parse_args()
+
+        return jsonable_encoder({"success": PluginAutoUpgradeService.exclude_plugin(tenant_id, args["plugin_id"])})
+
+
 api.add_resource(PluginDebuggingKeyApi, "/workspaces/current/plugin/debugging-key")
 api.add_resource(PluginListApi, "/workspaces/current/plugin/list")
 api.add_resource(PluginListLatestVersionsApi, "/workspaces/current/plugin/list/latest-versions")
@@ -521,3 +667,9 @@ api.add_resource(PluginFetchMarketplacePkgApi, "/workspaces/current/plugin/marke

 api.add_resource(PluginChangePermissionApi, "/workspaces/current/plugin/permission/change")
 api.add_resource(PluginFetchPermissionApi, "/workspaces/current/plugin/permission/fetch")
+
+api.add_resource(PluginFetchDynamicSelectOptionsApi, "/workspaces/current/plugin/parameters/dynamic-options")
+
+api.add_resource(PluginFetchPreferencesApi, "/workspaces/current/plugin/preferences/fetch")
+api.add_resource(PluginChangePreferencesApi, "/workspaces/current/plugin/preferences/change")
+api.add_resource(PluginAutoUpgradeExcludePluginApi, "/workspaces/current/plugin/preferences/autoupgrade/exclude")
@@ -1,6 +1,7 @@
 import io
+from urllib.parse import urlparse

-from flask import send_file
+from flask import redirect, send_file
 from flask_login import current_user
 from flask_restful import Resource, reqparse
 from sqlalchemy.orm import Session
@@ -9,17 +10,34 @@ from werkzeug.exceptions import Forbidden
 from configs import dify_config
 from controllers.console import api
 from controllers.console.wraps import account_initialization_required, enterprise_license_required, setup_required
+from core.mcp.auth.auth_flow import auth, handle_callback
+from core.mcp.auth.auth_provider import OAuthClientProvider
+from core.mcp.error import MCPAuthError, MCPError
+from core.mcp.mcp_client import MCPClient
 from core.model_runtime.utils.encoders import jsonable_encoder
 from extensions.ext_database import db
 from libs.helper import alphanumeric, uuid_value
 from libs.login import login_required
 from services.tools.api_tools_manage_service import ApiToolManageService
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService
+from services.tools.mcp_tools_mange_service import MCPToolManageService
 from services.tools.tool_labels_service import ToolLabelsService
 from services.tools.tools_manage_service import ToolCommonService
+from services.tools.tools_transform_service import ToolTransformService
 from services.tools.workflow_tools_manage_service import WorkflowToolManageService


+def is_valid_url(url: str) -> bool:
+    if not url:
+        return False
+
+    try:
+        parsed = urlparse(url)
+        return all([parsed.scheme, parsed.netloc]) and parsed.scheme in ["http", "https"]
+    except Exception:
+        return False
+
+
 class ToolProviderListApi(Resource):
    @setup_required
    @login_required
@@ -34,7 +52,7 @@ class ToolProviderListApi(Resource):
        req.add_argument(
            "type",
            type=str,
-            choices=["builtin", "model", "api", "workflow"],
+            choices=["builtin", "model", "api", "workflow", "mcp"],
            required=False,
            nullable=True,
            location="args",
@@ -613,6 +631,166 @@ class ToolLabelsApi(Resource):
        return jsonable_encoder(ToolLabelsService.list_tool_labels())


+class ToolProviderMCPApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("server_url", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("name", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("icon", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("icon_type", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json", default="")
+        parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        user = current_user
+        if not is_valid_url(args["server_url"]):
+            raise ValueError("Server URL is not valid.")
+        return jsonable_encoder(
+            MCPToolManageService.create_mcp_provider(
+                tenant_id=user.current_tenant_id,
+                server_url=args["server_url"],
+                name=args["name"],
+                icon=args["icon"],
+                icon_type=args["icon_type"],
+                icon_background=args["icon_background"],
+                user_id=user.id,
+                server_identifier=args["server_identifier"],
+            )
+        )
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def put(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("server_url", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("name", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("icon", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("icon_type", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("icon_background", type=str, required=False, nullable=True, location="json")
+        parser.add_argument("provider_id", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("server_identifier", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        if not is_valid_url(args["server_url"]):
+            if "[__HIDDEN__]" in args["server_url"]:
+                pass
+            else:
+                raise ValueError("Server URL is not valid.")
+        MCPToolManageService.update_mcp_provider(
+            tenant_id=current_user.current_tenant_id,
+            provider_id=args["provider_id"],
+            server_url=args["server_url"],
+            name=args["name"],
+            icon=args["icon"],
+            icon_type=args["icon_type"],
+            icon_background=args["icon_background"],
+            server_identifier=args["server_identifier"],
+        )
+        return {"result": "success"}
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def delete(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("provider_id", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+        MCPToolManageService.delete_mcp_tool(tenant_id=current_user.current_tenant_id, provider_id=args["provider_id"])
+        return {"result": "success"}
+
+
+class ToolMCPAuthApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("provider_id", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("authorization_code", type=str, required=False, nullable=True, location="json")
+        args = parser.parse_args()
+        provider_id = args["provider_id"]
+        tenant_id = current_user.current_tenant_id
+        provider = MCPToolManageService.get_mcp_provider_by_provider_id(provider_id, tenant_id)
+        if not provider:
+            raise ValueError("provider not found")
+        try:
+            with MCPClient(
+                provider.decrypted_server_url,
+                provider_id,
+                tenant_id,
+                authed=False,
+                authorization_code=args["authorization_code"],
+                for_list=True,
+            ):
+                MCPToolManageService.update_mcp_provider_credentials(
+                    mcp_provider=provider,
+                    credentials=provider.decrypted_credentials,
+                    authed=True,
+                )
+                return {"result": "success"}
+
+        except MCPAuthError:
+            auth_provider = OAuthClientProvider(provider_id, tenant_id, for_list=True)
+            return auth(auth_provider, provider.decrypted_server_url, args["authorization_code"])
+        except MCPError as e:
+            MCPToolManageService.update_mcp_provider_credentials(
+                mcp_provider=provider,
+                credentials={},
+                authed=False,
+            )
+            raise ValueError(f"Failed to connect to MCP server: {e}") from e
+
+
+class ToolMCPDetailApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider_id):
+        user = current_user
+        provider = MCPToolManageService.get_mcp_provider_by_provider_id(provider_id, user.current_tenant_id)
+        return jsonable_encoder(ToolTransformService.mcp_provider_to_user_provider(provider, for_list=True))
+
+
+class ToolMCPListAllApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self):
+        user = current_user
+        tenant_id = user.current_tenant_id
+
+        tools = MCPToolManageService.retrieve_mcp_tools(tenant_id=tenant_id)
+
+        return [tool.to_dict() for tool in tools]
+
+
+class ToolMCPUpdateApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, provider_id):
+        tenant_id = current_user.current_tenant_id
+        tools = MCPToolManageService.list_mcp_tool_from_remote_server(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+        )
+        return jsonable_encoder(tools)
+
+
+class ToolMCPCallbackApi(Resource):
+    def get(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("code", type=str, required=True, nullable=False, location="args")
+        parser.add_argument("state", type=str, required=True, nullable=False, location="args")
+        args = parser.parse_args()
+        state_key = args["state"]
+        authorization_code = args["code"]
+        handle_callback(state_key, authorization_code)
+        return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
+
+
 # tool provider
 api.add_resource(ToolProviderListApi, "/workspaces/current/tool-providers")

@@ -647,8 +825,15 @@ api.add_resource(ToolWorkflowProviderDeleteApi, "/workspaces/current/tool-provid
 api.add_resource(ToolWorkflowProviderGetApi, "/workspaces/current/tool-provider/workflow/get")
 api.add_resource(ToolWorkflowProviderListToolApi, "/workspaces/current/tool-provider/workflow/tools")

+# mcp tool provider
+api.add_resource(ToolMCPDetailApi, "/workspaces/current/tool-provider/mcp/tools/<path:provider_id>")
+api.add_resource(ToolProviderMCPApi, "/workspaces/current/tool-provider/mcp")
+api.add_resource(ToolMCPUpdateApi, "/workspaces/current/tool-provider/mcp/update/<path:provider_id>")
+api.add_resource(ToolMCPAuthApi, "/workspaces/current/tool-provider/mcp/auth")
+api.add_resource(ToolMCPCallbackApi, "/mcp/oauth/callback")
+
 api.add_resource(ToolBuiltinListApi, "/workspaces/current/tools/builtin")
 api.add_resource(ToolApiListApi, "/workspaces/current/tools/api")
+api.add_resource(ToolMCPListAllApi, "/workspaces/current/tools/mcp")
 api.add_resource(ToolWorkflowListApi, "/workspaces/current/tools/workflow")
-
 api.add_resource(ToolLabelsApi, "/workspaces/current/tool-labels")
@@ -87,7 +87,5 @@ class PluginUploadFileApi(Resource):
        except services.errors.file.UnsupportedFileTypeError:
            raise UnsupportedFileTypeError()

-        return tool_file, 201
-

 api.add_resource(PluginUploadFileApi, "/files/upload/for-plugin")
@@ -17,6 +17,7 @@ from core.plugin.entities.request import (
    RequestInvokeApp,
    RequestInvokeEncrypt,
    RequestInvokeLLM,
+    RequestInvokeLLMWithStructuredOutput,
    RequestInvokeModeration,
    RequestInvokeParameterExtractorNode,
    RequestInvokeQuestionClassifierNode,
@@ -47,6 +48,21 @@ class PluginInvokeLLMApi(Resource):
        return length_prefixed_response(0xF, generator())


+class PluginInvokeLLMWithStructuredOutputApi(Resource):
+    @setup_required
+    @plugin_inner_api_only
+    @get_user_tenant
+    @plugin_data(payload_type=RequestInvokeLLMWithStructuredOutput)
+    def post(self, user_model: Account | EndUser, tenant_model: Tenant, payload: RequestInvokeLLMWithStructuredOutput):
+        def generator():
+            response = PluginModelBackwardsInvocation.invoke_llm_with_structured_output(
+                user_model.id, tenant_model, payload
+            )
+            return PluginModelBackwardsInvocation.convert_to_event_stream(response)
+
+        return length_prefixed_response(0xF, generator())
+
+
 class PluginInvokeTextEmbeddingApi(Resource):
    @setup_required
    @plugin_inner_api_only
@@ -291,6 +307,7 @@ class PluginFetchAppInfoApi(Resource):


 api.add_resource(PluginInvokeLLMApi, "/invoke/llm")
+api.add_resource(PluginInvokeLLMWithStructuredOutputApi, "/invoke/llm/structured-output")
 api.add_resource(PluginInvokeTextEmbeddingApi, "/invoke/text-embedding")
 api.add_resource(PluginInvokeRerankApi, "/invoke/rerank")
 api.add_resource(PluginInvokeTTSApi, "/invoke/tts")
@@ -29,7 +29,19 @@ class EnterpriseWorkspace(Resource):

        tenant_was_created.send(tenant)

-        return {"message": "enterprise workspace created."}
+        resp = {
+            "id": tenant.id,
+            "name": tenant.name,
+            "plan": tenant.plan,
+            "status": tenant.status,
+            "created_at": tenant.created_at.isoformat() + "Z" if tenant.created_at else None,
+            "updated_at": tenant.updated_at.isoformat() + "Z" if tenant.updated_at else None,
+        }
+
+        return {
+            "message": "enterprise workspace created.",
+            "tenant": resp,
+        }


 class EnterpriseWorkspaceNoOwnerEmail(Resource):
@@ -0,0 +1,8 @@
+from flask import Blueprint
+
+from libs.external_api import ExternalApi
+
+bp = Blueprint("mcp", __name__, url_prefix="/mcp")
+api = ExternalApi(bp)
+
+from . import mcp
@@ -0,0 +1,104 @@
+from flask_restful import Resource, reqparse
+from pydantic import ValidationError
+
+from controllers.console.app.mcp_server import AppMCPServerStatus
+from controllers.mcp import api
+from core.app.app_config.entities import VariableEntity
+from core.mcp import types
+from core.mcp.server.streamable_http import MCPServerStreamableHTTPRequestHandler
+from core.mcp.types import ClientNotification, ClientRequest
+from core.mcp.utils import create_mcp_error_response
+from extensions.ext_database import db
+from libs import helper
+from models.model import App, AppMCPServer, AppMode
+
+
+class MCPAppApi(Resource):
+    def post(self, server_code):
+        def int_or_str(value):
+            if isinstance(value, (int, str)):
+                return value
+            else:
+                return None
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("jsonrpc", type=str, required=True, location="json")
+        parser.add_argument("method", type=str, required=True, location="json")
+        parser.add_argument("params", type=dict, required=False, location="json")
+        parser.add_argument("id", type=int_or_str, required=False, location="json")
+        args = parser.parse_args()
+
+        request_id = args.get("id")
+
+        server = db.session.query(AppMCPServer).filter(AppMCPServer.server_code == server_code).first()
+        if not server:
+            return helper.compact_generate_response(
+                create_mcp_error_response(request_id, types.INVALID_REQUEST, "Server Not Found")
+            )
+
+        if server.status != AppMCPServerStatus.ACTIVE:
+            return helper.compact_generate_response(
+                create_mcp_error_response(request_id, types.INVALID_REQUEST, "Server is not active")
+            )
+
+        app = db.session.query(App).filter(App.id == server.app_id).first()
+        if not app:
+            return helper.compact_generate_response(
+                create_mcp_error_response(request_id, types.INVALID_REQUEST, "App Not Found")
+            )
+
+        if app.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}:
+            workflow = app.workflow
+            if workflow is None:
+                return helper.compact_generate_response(
+                    create_mcp_error_response(request_id, types.INVALID_REQUEST, "App is unavailable")
+                )
+
+            user_input_form = workflow.user_input_form(to_old_structure=True)
+        else:
+            app_model_config = app.app_model_config
+            if app_model_config is None:
+                return helper.compact_generate_response(
+                    create_mcp_error_response(request_id, types.INVALID_REQUEST, "App is unavailable")
+                )
+
+            features_dict = app_model_config.to_dict()
+            user_input_form = features_dict.get("user_input_form", [])
+        converted_user_input_form: list[VariableEntity] = []
+        try:
+            for item in user_input_form:
+                variable_type = item.get("type", "") or list(item.keys())[0]
+                variable = item[variable_type]
+                converted_user_input_form.append(
+                    VariableEntity(
+                        type=variable_type,
+                        variable=variable.get("variable"),
+                        description=variable.get("description") or "",
+                        label=variable.get("label"),
+                        required=variable.get("required", False),
+                        max_length=variable.get("max_length"),
+                        options=variable.get("options") or [],
+                    )
+                )
+        except ValidationError as e:
+            return helper.compact_generate_response(
+                create_mcp_error_response(request_id, types.INVALID_PARAMS, f"Invalid user_input_form: {str(e)}")
+            )
+
+        try:
+            request: ClientRequest | ClientNotification = ClientRequest.model_validate(args)
+        except ValidationError as e:
+            try:
+                notification = ClientNotification.model_validate(args)
+                request = notification
+            except ValidationError as e:
+                return helper.compact_generate_response(
+                    create_mcp_error_response(request_id, types.INVALID_PARAMS, f"Invalid MCP request: {str(e)}")
+                )
+
+        mcp_server_handler = MCPServerStreamableHTTPRequestHandler(app, request, converted_user_input_form)
+        response = mcp_server_handler.handle()
+        return helper.compact_generate_response(response)
+
+
+api.add_resource(MCPAppApi, "/server/<string:server_code>/mcp")
@@ -20,7 +20,7 @@ from controllers.service_api.app.error import (
 from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeError
-from models.model import App, AppMode, EndUser
+from models.model import App, EndUser
 from services.audio_service import AudioService
 from services.errors.audio import (
    AudioTooLargeServiceError,
@@ -78,20 +78,9 @@ class TextApi(Resource):

            message_id = args.get("message_id", None)
            text = args.get("text", None)
-            if (
-                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
-                and app_model.workflow
-                and app_model.workflow.features_dict
-            ):
-                text_to_speech = app_model.workflow.features_dict.get("text_to_speech", {})
-                voice = args.get("voice") or text_to_speech.get("voice")
-            else:
-                try:
-                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
-                except Exception:
-                    voice = None
+            voice = args.get("voice", None)
            response = AudioService.transcript_tts(
-                app_model=app_model, message_id=message_id, end_user=end_user.external_user_id, voice=voice, text=text
+                app_model=app_model, text=text, voice=voice, end_user=end_user.external_user_id, message_id=message_id
            )

            return response
@@ -133,6 +133,22 @@ class DatasetListApi(DatasetApiResource):
        parser.add_argument("embedding_model_provider", type=str, required=False, nullable=True, location="json")

        args = parser.parse_args()
+
+        if args.get("embedding_model_provider"):
+            DatasetService.check_embedding_model_setting(
+                tenant_id, args.get("embedding_model_provider"), args.get("embedding_model")
+            )
+        if (
+            args.get("retrieval_model")
+            and args.get("retrieval_model").get("reranking_model")
+            and args.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+        ):
+            DatasetService.check_reranking_model_setting(
+                tenant_id,
+                args.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
+                args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+            )
+
        try:
            dataset = DatasetService.create_empty_dataset(
                tenant_id=tenant_id,
@@ -265,10 +281,20 @@ class DatasetApi(DatasetApiResource):
        data = request.get_json()

        # check embedding model setting
-        if data.get("indexing_technique") == "high_quality":
+        if data.get("indexing_technique") == "high_quality" or data.get("embedding_model_provider"):
            DatasetService.check_embedding_model_setting(
                dataset.tenant_id, data.get("embedding_model_provider"), data.get("embedding_model")
            )
+        if (
+            data.get("retrieval_model")
+            and data.get("retrieval_model").get("reranking_model")
+            and data.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+        ):
+            DatasetService.check_reranking_model_setting(
+                dataset.tenant_id,
+                data.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
+                data.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+            )

        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
        DatasetPermissionService.check_permission(
@@ -3,7 +3,7 @@ import json
 from flask import request
 from flask_restful import marshal, reqparse
 from sqlalchemy import desc, select
-from werkzeug.exceptions import NotFound
+from werkzeug.exceptions import Forbidden, NotFound

 import services
 from controllers.common.errors import FilenameNotExistsError
@@ -18,6 +18,7 @@ from controllers.service_api.app.error import (
 from controllers.service_api.dataset.error import (
    ArchivedDocumentImmutableError,
    DocumentIndexingError,
+    InvalidMetadataError,
 )
 from controllers.service_api.wraps import (
    DatasetApiResource,
@@ -29,7 +30,7 @@ from extensions.ext_database import db
 from fields.document_fields import document_fields, document_status_fields
 from libs.login import current_user
 from models.dataset import Dataset, Document, DocumentSegment
-from services.dataset_service import DocumentService
+from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig
 from services.file_service import FileService

@@ -59,6 +60,7 @@ class DocumentAddByTextApi(DatasetApiResource):
        parser.add_argument("embedding_model_provider", type=str, required=False, nullable=True, location="json")

        args = parser.parse_args()
+
        dataset_id = str(dataset_id)
        tenant_id = str(tenant_id)
        dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
@@ -74,6 +76,21 @@ class DocumentAddByTextApi(DatasetApiResource):
        if text is None or name is None:
            raise ValueError("Both 'text' and 'name' must be non-null values.")

+        if args.get("embedding_model_provider"):
+            DatasetService.check_embedding_model_setting(
+                tenant_id, args.get("embedding_model_provider"), args.get("embedding_model")
+            )
+        if (
+            args.get("retrieval_model")
+            and args.get("retrieval_model").get("reranking_model")
+            and args.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+        ):
+            DatasetService.check_reranking_model_setting(
+                tenant_id,
+                args.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
+                args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+            )
+
        upload_file = FileService.upload_text(text=str(text), text_name=str(name))
        data_source = {
            "type": "upload_file",
@@ -124,6 +141,17 @@ class DocumentUpdateByTextApi(DatasetApiResource):
        if not dataset:
            raise ValueError("Dataset does not exist.")

+        if (
+            args.get("retrieval_model")
+            and args.get("retrieval_model").get("reranking_model")
+            and args.get("retrieval_model").get("reranking_model").get("reranking_provider_name")
+        ):
+            DatasetService.check_reranking_model_setting(
+                tenant_id,
+                args.get("retrieval_model").get("reranking_model").get("reranking_provider_name"),
+                args.get("retrieval_model").get("reranking_model").get("reranking_model_name"),
+            )
+
        # indexing_technique is already set in dataset since this is an update
        args["indexing_technique"] = dataset.indexing_technique

@@ -183,11 +211,29 @@ class DocumentAddByFileApi(DatasetApiResource):
        if not dataset:
            raise ValueError("Dataset does not exist.")

+        if dataset.provider == "external":
+            raise ValueError("External datasets are not supported.")
+
        indexing_technique = args.get("indexing_technique") or dataset.indexing_technique
        if not indexing_technique:
            raise ValueError("indexing_technique is required.")
        args["indexing_technique"] = indexing_technique

+        if "embedding_model_provider" in args:
+            DatasetService.check_embedding_model_setting(
+                tenant_id, args["embedding_model_provider"], args["embedding_model"]
+            )
+        if (
+            "retrieval_model" in args
+            and args["retrieval_model"].get("reranking_model")
+            and args["retrieval_model"].get("reranking_model").get("reranking_provider_name")
+        ):
+            DatasetService.check_reranking_model_setting(
+                tenant_id,
+                args["retrieval_model"].get("reranking_model").get("reranking_provider_name"),
+                args["retrieval_model"].get("reranking_model").get("reranking_model_name"),
+            )
+
        # save file info
        file = request.files["file"]
        # check file
@@ -258,6 +304,9 @@ class DocumentUpdateByFileApi(DatasetApiResource):
        if not dataset:
            raise ValueError("Dataset does not exist.")

+        if dataset.provider == "external":
+            raise ValueError("External datasets are not supported.")
+
        # indexing_technique is already set in dataset since this is an update
        args["indexing_technique"] = dataset.indexing_technique

@@ -424,6 +473,101 @@ class DocumentIndexingStatusApi(DatasetApiResource):
        return data


+class DocumentDetailApi(DatasetApiResource):
+    METADATA_CHOICES = {"all", "only", "without"}
+
+    def get(self, tenant_id, dataset_id, document_id):
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+
+        dataset = self.get_dataset(dataset_id, tenant_id)
+
+        document = DocumentService.get_document(dataset.id, document_id)
+
+        if not document:
+            raise NotFound("Document not found.")
+
+        if document.tenant_id != str(tenant_id):
+            raise Forbidden("No permission.")
+
+        metadata = request.args.get("metadata", "all")
+        if metadata not in self.METADATA_CHOICES:
+            raise InvalidMetadataError(f"Invalid metadata value: {metadata}")
+
+        if metadata == "only":
+            response = {"id": document.id, "doc_type": document.doc_type, "doc_metadata": document.doc_metadata_details}
+        elif metadata == "without":
+            dataset_process_rules = DatasetService.get_process_rules(dataset_id)
+            document_process_rules = document.dataset_process_rule.to_dict()
+            data_source_info = document.data_source_detail_dict
+            response = {
+                "id": document.id,
+                "position": document.position,
+                "data_source_type": document.data_source_type,
+                "data_source_info": data_source_info,
+                "dataset_process_rule_id": document.dataset_process_rule_id,
+                "dataset_process_rule": dataset_process_rules,
+                "document_process_rule": document_process_rules,
+                "name": document.name,
+                "created_from": document.created_from,
+                "created_by": document.created_by,
+                "created_at": document.created_at.timestamp(),
+                "tokens": document.tokens,
+                "indexing_status": document.indexing_status,
+                "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
+                "updated_at": int(document.updated_at.timestamp()) if document.updated_at else None,
+                "indexing_latency": document.indexing_latency,
+                "error": document.error,
+                "enabled": document.enabled,
+                "disabled_at": int(document.disabled_at.timestamp()) if document.disabled_at else None,
+                "disabled_by": document.disabled_by,
+                "archived": document.archived,
+                "segment_count": document.segment_count,
+                "average_segment_length": document.average_segment_length,
+                "hit_count": document.hit_count,
+                "display_status": document.display_status,
+                "doc_form": document.doc_form,
+                "doc_language": document.doc_language,
+            }
+        else:
+            dataset_process_rules = DatasetService.get_process_rules(dataset_id)
+            document_process_rules = document.dataset_process_rule.to_dict()
+            data_source_info = document.data_source_detail_dict
+            response = {
+                "id": document.id,
+                "position": document.position,
+                "data_source_type": document.data_source_type,
+                "data_source_info": data_source_info,
+                "dataset_process_rule_id": document.dataset_process_rule_id,
+                "dataset_process_rule": dataset_process_rules,
+                "document_process_rule": document_process_rules,
+                "name": document.name,
+                "created_from": document.created_from,
+                "created_by": document.created_by,
+                "created_at": document.created_at.timestamp(),
+                "tokens": document.tokens,
+                "indexing_status": document.indexing_status,
+                "completed_at": int(document.completed_at.timestamp()) if document.completed_at else None,
+                "updated_at": int(document.updated_at.timestamp()) if document.updated_at else None,
+                "indexing_latency": document.indexing_latency,
+                "error": document.error,
+                "enabled": document.enabled,
+                "disabled_at": int(document.disabled_at.timestamp()) if document.disabled_at else None,
+                "disabled_by": document.disabled_by,
+                "archived": document.archived,
+                "doc_type": document.doc_type,
+                "doc_metadata": document.doc_metadata_details,
+                "segment_count": document.segment_count,
+                "average_segment_length": document.average_segment_length,
+                "hit_count": document.hit_count,
+                "display_status": document.display_status,
+                "doc_form": document.doc_form,
+                "doc_language": document.doc_language,
+            }
+
+        return response
+
+
 api.add_resource(
    DocumentAddByTextApi,
    "/datasets/<uuid:dataset_id>/document/create_by_text",
@@ -447,3 +591,4 @@ api.add_resource(
 api.add_resource(DocumentDeleteApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
 api.add_resource(DocumentListApi, "/datasets/<uuid:dataset_id>/documents")
 api.add_resource(DocumentIndexingStatusApi, "/datasets/<uuid:dataset_id>/documents/<string:batch>/indexing-status")
+api.add_resource(DocumentDetailApi, "/datasets/<uuid:dataset_id>/documents/<uuid:document_id>")
@@ -9,7 +9,7 @@ class IndexApi(Resource):
        return {
            "welcome": "Dify OpenAPI",
            "api_version": "v1",
-            "server_version": dify_config.CURRENT_VERSION,
+            "server_version": dify_config.project.version,
        }


@@ -11,13 +11,13 @@ from flask_restful import Resource
 from pydantic import BaseModel
 from sqlalchemy import select, update
 from sqlalchemy.orm import Session
-from werkzeug.exceptions import Forbidden, Unauthorized
+from werkzeug.exceptions import Forbidden, NotFound, Unauthorized

 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from libs.login import _get_user
 from models.account import Account, Tenant, TenantAccountJoin, TenantStatus
-from models.dataset import RateLimitLog
+from models.dataset import Dataset, RateLimitLog
 from models.model import ApiToken, App, EndUser
 from services.feature_service import FeatureService

@@ -317,3 +317,11 @@ def create_or_update_end_user_for_user_id(app_model: App, user_id: Optional[str]

 class DatasetApiResource(Resource):
    method_decorators = [validate_dataset_token]
+
+    def get_dataset(self, dataset_id: str, tenant_id: str) -> Dataset:
+        dataset = db.session.query(Dataset).filter(Dataset.id == dataset_id, Dataset.tenant_id == tenant_id).first()
+
+        if not dataset:
+            raise NotFound("Dataset not found.")
+
+        return dataset
@@ -19,7 +19,7 @@ from controllers.web.error import (
 from controllers.web.wraps import WebApiResource
 from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeError
-from models.model import App, AppMode
+from models.model import App
 from services.audio_service import AudioService
 from services.errors.audio import (
    AudioTooLargeServiceError,
@@ -77,21 +77,9 @@ class TextApi(WebApiResource):

            message_id = args.get("message_id", None)
            text = args.get("text", None)
-            if (
-                app_model.mode in {AppMode.ADVANCED_CHAT.value, AppMode.WORKFLOW.value}
-                and app_model.workflow
-                and app_model.workflow.features_dict
-            ):
-                text_to_speech = app_model.workflow.features_dict.get("text_to_speech", {})
-                voice = args.get("voice") or text_to_speech.get("voice")
-            else:
-                try:
-                    voice = args.get("voice") or app_model.app_model_config.text_to_speech_dict.get("voice")
-                except Exception:
-                    voice = None
-
+            voice = args.get("voice", None)
            response = AudioService.transcript_tts(
-                app_model=app_model, message_id=message_id, end_user=end_user.external_user_id, voice=voice, text=text
+                app_model=app_model, text=text, voice=voice, end_user=end_user.external_user_id, message_id=message_id
            )

            return response
@@ -139,3 +139,13 @@ class InvokeRateLimitError(BaseHTTPException):
    error_code = "rate_limit_error"
    description = "Rate Limit Error"
    code = 429
+
+
+class NotFoundError(BaseHTTPException):
+    error_code = "not_found"
+    code = 404
+
+
+class InvalidArgumentError(BaseHTTPException):
+    error_code = "invalid_param"
+    code = 400
@@ -161,10 +161,14 @@ class BaseAgentRunner(AppRunner):
            if parameter.type == ToolParameter.ToolParameterType.SELECT:
                enum = [option.value for option in parameter.options] if parameter.options else []

-            message_tool.parameters["properties"][parameter.name] = {
-                "type": parameter_type,
-                "description": parameter.llm_description or "",
-            }
+            message_tool.parameters["properties"][parameter.name] = (
+                {
+                    "type": parameter_type,
+                    "description": parameter.llm_description or "",
+                }
+                if parameter.input_schema is None
+                else parameter.input_schema
+            )

            if len(enum) > 0:
                message_tool.parameters["properties"][parameter.name]["enum"] = enum
@@ -254,10 +258,14 @@ class BaseAgentRunner(AppRunner):
            if parameter.type == ToolParameter.ToolParameterType.SELECT:
                enum = [option.value for option in parameter.options] if parameter.options else []

-            prompt_tool.parameters["properties"][parameter.name] = {
-                "type": parameter_type,
-                "description": parameter.llm_description or "",
-            }
+            prompt_tool.parameters["properties"][parameter.name] = (
+                {
+                    "type": parameter_type,
+                    "description": parameter.llm_description or "",
+                }
+                if parameter.input_schema is None
+                else parameter.input_schema
+            )

            if len(enum) > 0:
                prompt_tool.parameters["properties"][parameter.name]["enum"] = enum
@@ -85,7 +85,7 @@ class AgentStrategyEntity(BaseModel):
    description: I18nObject = Field(..., description="The description of the agent strategy")
    output_schema: Optional[dict] = None
    features: Optional[list[AgentFeature]] = None
-
+    meta_version: Optional[str] = None
    # pydantic configs
    model_config = ConfigDict(protected_namespaces=())

@@ -15,10 +15,12 @@ class PluginAgentStrategy(BaseAgentStrategy):

    tenant_id: str
    declaration: AgentStrategyEntity
+    meta_version: str | None = None

-    def __init__(self, tenant_id: str, declaration: AgentStrategyEntity):
+    def __init__(self, tenant_id: str, declaration: AgentStrategyEntity, meta_version: str | None):
        self.tenant_id = tenant_id
        self.declaration = declaration
+        self.meta_version = meta_version

    def get_parameters(self) -> Sequence[AgentStrategyParameter]:
        return self.declaration.parameters
@@ -104,6 +104,7 @@ class VariableEntity(BaseModel):
    Variable Entity.
    """

+    # `variable` records the name of the variable in user inputs.
    variable: str
    label: str
    description: str = ""
@@ -27,15 +27,22 @@ from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
 from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
+from core.workflow.repositories.draft_variable_repository import (
+    DraftVariableSaverFactory,
+)
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from extensions.ext_database import db
 from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.enums import WorkflowRunTriggeredFrom
 from services.conversation_service import ConversationService
-from services.errors.message import MessageNotExistsError
+from services.workflow_draft_variable_service import (
+    DraftVarLoader,
+    WorkflowDraftVariableService,
+)

 logger = logging.getLogger(__name__)

@@ -116,6 +123,11 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            )

        # parse files
+        # TODO(QuantumGhost): Move file parsing logic to the API controller layer
+        # for better separation of concerns.
+        #
+        # For implementation reference, see the `_parse_file` function and
+        # `DraftWorkflowNodeRunApi` class which handle this properly.
        files = args["files"] if args.get("files") else []
        file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
        if file_extra_config:
@@ -261,6 +273,13 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            app_id=application_generate_entity.app_config.app_id,
            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )
+        var_loader = DraftVarLoader(
+            engine=db.engine,
+            app_id=application_generate_entity.app_config.app_id,
+            tenant_id=application_generate_entity.app_config.tenant_id,
+        )
+        draft_var_srv = WorkflowDraftVariableService(db.session())
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)

        return self._generate(
            workflow=workflow,
@@ -271,6 +290,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            workflow_node_execution_repository=workflow_node_execution_repository,
            conversation=None,
            stream=streaming,
+            variable_loader=var_loader,
        )

    def single_loop_generate(
@@ -336,6 +356,13 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            app_id=application_generate_entity.app_config.app_id,
            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )
+        var_loader = DraftVarLoader(
+            engine=db.engine,
+            app_id=application_generate_entity.app_config.app_id,
+            tenant_id=application_generate_entity.app_config.tenant_id,
+        )
+        draft_var_srv = WorkflowDraftVariableService(db.session())
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)

        return self._generate(
            workflow=workflow,
@@ -346,6 +373,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            workflow_node_execution_repository=workflow_node_execution_repository,
            conversation=None,
            stream=streaming,
+            variable_loader=var_loader,
        )

    def _generate(
@@ -359,6 +387,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
        conversation: Optional[Conversation] = None,
        stream: bool = True,
+        variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
    ) -> Mapping[str, Any] | Generator[str | Mapping[str, Any], Any, None]:
        """
        Generate App response.
@@ -410,6 +439,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                "conversation_id": conversation.id,
                "message_id": message.id,
                "context": context,
+                "variable_loader": variable_loader,
            },
        )

@@ -426,6 +456,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            stream=stream,
+            draft_var_saver_factory=self._get_draft_var_saver_factory(invoke_from),
        )

        return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
@@ -438,6 +469,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
        conversation_id: str,
        message_id: str,
        context: contextvars.Context,
+        variable_loader: VariableLoader,
    ) -> None:
        """
        Generate worker in a new thread.
@@ -454,8 +486,6 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                # get conversation and message
                conversation = self._get_conversation(conversation_id)
                message = self._get_message(message_id)
-                if message is None:
-                    raise MessageNotExistsError("Message not exists")

                # chatbot app
                runner = AdvancedChatAppRunner(
@@ -464,6 +494,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                    conversation=conversation,
                    message=message,
                    dialogue_count=self._dialogue_count,
+                    variable_loader=variable_loader,
                )

                runner.run()
@@ -497,6 +528,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
        user: Union[Account, EndUser],
        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        draft_var_saver_factory: DraftVariableSaverFactory,
        stream: bool = False,
    ) -> Union[ChatbotAppBlockingResponse, Generator[ChatbotAppStreamResponse, None, None]]:
        """
@@ -523,6 +555,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            stream=stream,
+            draft_var_saver_factory=draft_var_saver_factory,
        )

        try:
@@ -19,6 +19,7 @@ from core.moderation.base import ModerationError
 from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
 from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.enums import SystemVariableKey
+from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from models.enums import UserFrom
@@ -40,14 +41,17 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
        conversation: Conversation,
        message: Message,
        dialogue_count: int,
+        variable_loader: VariableLoader,
    ) -> None:
-        super().__init__(queue_manager)
-
+        super().__init__(queue_manager, variable_loader)
        self.application_generate_entity = application_generate_entity
        self.conversation = conversation
        self.message = message
        self._dialogue_count = dialogue_count

+    def _get_app_id(self) -> str:
+        return self.application_generate_entity.app_config.app_id
+
    def run(self) -> None:
        app_config = self.application_generate_entity.app_config
        app_config = cast(AdvancedChatAppConfig, app_config)
@@ -64,6 +64,7 @@ from core.workflow.entities.workflow_execution import WorkflowExecutionStatus, W
 from core.workflow.enums import SystemVariableKey
 from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
+from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
@@ -94,6 +95,7 @@ class AdvancedChatAppGenerateTaskPipeline:
        dialogue_count: int,
        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        draft_var_saver_factory: DraftVariableSaverFactory,
    ) -> None:
        self._base_task_pipeline = BasedGenerateTaskPipeline(
            application_generate_entity=application_generate_entity,
@@ -153,6 +155,7 @@ class AdvancedChatAppGenerateTaskPipeline:
        self._conversation_name_generate_thread: Thread | None = None
        self._recorded_files: list[Mapping[str, Any]] = []
        self._workflow_run_id: str = ""
+        self._draft_var_saver_factory = draft_var_saver_factory

    def process(self) -> Union[ChatbotAppBlockingResponse, Generator[ChatbotAppStreamResponse, None, None]]:
        """
@@ -371,6 +374,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                        workflow_node_execution=workflow_node_execution,
                    )
                    session.commit()
+                self._save_output_for_event(event, workflow_node_execution.id)

                if node_finish_resp:
                    yield node_finish_resp
@@ -390,6 +394,8 @@ class AdvancedChatAppGenerateTaskPipeline:
                    task_id=self._application_generate_entity.task_id,
                    workflow_node_execution=workflow_node_execution,
                )
+                if isinstance(event, QueueNodeExceptionEvent):
+                    self._save_output_for_event(event, workflow_node_execution.id)

                if node_finish_resp:
                    yield node_finish_resp
@@ -759,3 +765,15 @@ class AdvancedChatAppGenerateTaskPipeline:
        if not message:
            raise ValueError(f"Message not found: {self._message_id}")
        return message
+
+    def _save_output_for_event(self, event: QueueNodeSucceededEvent | QueueNodeExceptionEvent, node_execution_id: str):
+        with Session(db.engine) as session, session.begin():
+            saver = self._draft_var_saver_factory(
+                session=session,
+                app_id=self._application_generate_entity.app_config.app_id,
+                node_id=event.node_id,
+                node_type=event.node_type,
+                node_execution_id=node_execution_id,
+                enclosing_node_id=event.in_loop_id or event.in_iteration_id,
+            )
+            saver.save(event.process_data, event.outputs)
@@ -26,7 +26,6 @@ from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, EndUser
 from services.conversation_service import ConversationService
-from services.errors.message import MessageNotExistsError

 logger = logging.getLogger(__name__)

@@ -124,6 +123,11 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
            override_model_config_dict["retriever_resource"] = {"enabled": True}

        # parse files
+        # TODO(QuantumGhost): Move file parsing logic to the API controller layer
+        # for better separation of concerns.
+        #
+        # For implementation reference, see the `_parse_file` function and
+        # `DraftWorkflowNodeRunApi` class which handle this properly.
        files = args.get("files") or []
        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
        if file_extra_config:
@@ -233,8 +237,6 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
                # get conversation and message
                conversation = self._get_conversation(conversation_id)
                message = self._get_message(message_id)
-                if message is None:
-                    raise MessageNotExistsError("Message not exists")

                # chatbot app
                runner = AgentChatAppRunner()
@@ -1,10 +1,20 @@
 import json
 from collections.abc import Generator, Mapping, Sequence
-from typing import TYPE_CHECKING, Any, Optional, Union
+from typing import TYPE_CHECKING, Any, Optional, Union, final
+
+from sqlalchemy.orm import Session

 from core.app.app_config.entities import VariableEntityType
+from core.app.entities.app_invoke_entities import InvokeFrom
 from core.file import File, FileUploadConfig
+from core.workflow.nodes.enums import NodeType
+from core.workflow.repositories.draft_variable_repository import (
+    DraftVariableSaver,
+    DraftVariableSaverFactory,
+    NoopDraftVariableSaver,
+)
 from factories import file_factory
+from services.workflow_draft_variable_service import DraftVariableSaver as DraftVariableSaverImpl

 if TYPE_CHECKING:
    from core.app.app_config.entities import VariableEntity
@@ -159,3 +169,38 @@ class BaseAppGenerator:
                        yield f"event: {message}\n\n"

            return gen()
+
+    @final
+    @staticmethod
+    def _get_draft_var_saver_factory(invoke_from: InvokeFrom) -> DraftVariableSaverFactory:
+        if invoke_from == InvokeFrom.DEBUGGER:
+
+            def draft_var_saver_factory(
+                session: Session,
+                app_id: str,
+                node_id: str,
+                node_type: NodeType,
+                node_execution_id: str,
+                enclosing_node_id: str | None = None,
+            ) -> DraftVariableSaver:
+                return DraftVariableSaverImpl(
+                    session=session,
+                    app_id=app_id,
+                    node_id=node_id,
+                    node_type=node_type,
+                    node_execution_id=node_execution_id,
+                    enclosing_node_id=enclosing_node_id,
+                )
+        else:
+
+            def draft_var_saver_factory(
+                session: Session,
+                app_id: str,
+                node_id: str,
+                node_type: NodeType,
+                node_execution_id: str,
+                enclosing_node_id: str | None = None,
+            ) -> DraftVariableSaver:
+                return NoopDraftVariableSaver()
+
+        return draft_var_saver_factory
@@ -25,7 +25,6 @@ from factories import file_factory
 from models.account import Account
 from models.model import App, EndUser
 from services.conversation_service import ConversationService
-from services.errors.message import MessageNotExistsError

 logger = logging.getLogger(__name__)

@@ -115,6 +114,11 @@ class ChatAppGenerator(MessageBasedAppGenerator):
            override_model_config_dict["retriever_resource"] = {"enabled": True}

        # parse files
+        # TODO(QuantumGhost): Move file parsing logic to the API controller layer
+        # for better separation of concerns.
+        #
+        # For implementation reference, see the `_parse_file` function and
+        # `DraftWorkflowNodeRunApi` class which handle this properly.
        files = args["files"] if args.get("files") else []
        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
        if file_extra_config:
@@ -219,8 +223,6 @@ class ChatAppGenerator(MessageBasedAppGenerator):
                # get conversation and message
                conversation = self._get_conversation(conversation_id)
                message = self._get_message(message_id)
-                if message is None:
-                    raise MessageNotExistsError("Message not exists")

                # chatbot app
                runner = ChatAppRunner()
@@ -44,10 +44,12 @@ from core.app.entities.task_entities import (
 )
 from core.file import FILE_MODEL_IDENTITY, File
 from core.tools.tool_manager import ToolManager
+from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.workflow.entities.workflow_execution import WorkflowExecution
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.tool.entities import ToolNodeData
+from core.workflow.workflow_type_encoder import WorkflowRuntimeTypeConverter
 from models import (
    Account,
    CreatorUserRole,
@@ -125,7 +127,7 @@ class WorkflowResponseConverter:
                id=workflow_execution.id_,
                workflow_id=workflow_execution.workflow_id,
                status=workflow_execution.status,
-                outputs=workflow_execution.outputs,
+                outputs=WorkflowRuntimeTypeConverter().to_json_encodable(workflow_execution.outputs),
                error=workflow_execution.error_message,
                elapsed_time=workflow_execution.elapsed_time,
                total_tokens=workflow_execution.total_tokens,
@@ -202,6 +204,8 @@ class WorkflowResponseConverter:
        if not workflow_node_execution.finished_at:
            return None

+        json_converter = WorkflowRuntimeTypeConverter()
+
        return NodeFinishStreamResponse(
            task_id=task_id,
            workflow_run_id=workflow_node_execution.workflow_execution_id,
@@ -214,7 +218,7 @@ class WorkflowResponseConverter:
                predecessor_node_id=workflow_node_execution.predecessor_node_id,
                inputs=workflow_node_execution.inputs,
                process_data=workflow_node_execution.process_data,
-                outputs=workflow_node_execution.outputs,
+                outputs=json_converter.to_json_encodable(workflow_node_execution.outputs),
                status=workflow_node_execution.status,
                error=workflow_node_execution.error,
                elapsed_time=workflow_node_execution.elapsed_time,
@@ -245,6 +249,8 @@ class WorkflowResponseConverter:
        if not workflow_node_execution.finished_at:
            return None

+        json_converter = WorkflowRuntimeTypeConverter()
+
        return NodeRetryStreamResponse(
            task_id=task_id,
            workflow_run_id=workflow_node_execution.workflow_execution_id,
@@ -257,7 +263,7 @@ class WorkflowResponseConverter:
                predecessor_node_id=workflow_node_execution.predecessor_node_id,
                inputs=workflow_node_execution.inputs,
                process_data=workflow_node_execution.process_data,
-                outputs=workflow_node_execution.outputs,
+                outputs=json_converter.to_json_encodable(workflow_node_execution.outputs),
                status=workflow_node_execution.status,
                error=workflow_node_execution.error,
                elapsed_time=workflow_node_execution.elapsed_time,
@@ -376,6 +382,7 @@ class WorkflowResponseConverter:
        workflow_execution_id: str,
        event: QueueIterationCompletedEvent,
    ) -> IterationNodeCompletedStreamResponse:
+        json_converter = WorkflowRuntimeTypeConverter()
        return IterationNodeCompletedStreamResponse(
            task_id=task_id,
            workflow_run_id=workflow_execution_id,
@@ -384,7 +391,7 @@ class WorkflowResponseConverter:
                node_id=event.node_id,
                node_type=event.node_type.value,
                title=event.node_data.title,
-                outputs=event.outputs,
+                outputs=json_converter.to_json_encodable(event.outputs),
                created_at=int(time.time()),
                extras={},
                inputs=event.inputs or {},
@@ -463,7 +470,7 @@ class WorkflowResponseConverter:
                node_id=event.node_id,
                node_type=event.node_type.value,
                title=event.node_data.title,
-                outputs=event.outputs,
+                outputs=WorkflowRuntimeTypeConverter().to_json_encodable(event.outputs),
                created_at=int(time.time()),
                extras={},
                inputs=event.inputs or {},
@@ -500,7 +507,8 @@ class WorkflowResponseConverter:
        # Convert to tuple to match Sequence type
        return tuple(flattened_files)

-    def _fetch_files_from_variable_value(self, value: Union[dict, list]) -> Sequence[Mapping[str, Any]]:
+    @classmethod
+    def _fetch_files_from_variable_value(cls, value: Union[dict, list, Segment]) -> Sequence[Mapping[str, Any]]:
        """
        Fetch files from variable value
        :param value: variable value
@@ -509,20 +517,30 @@ class WorkflowResponseConverter:
        if not value:
            return []

-        files = []
-        if isinstance(value, list):
+        files: list[Mapping[str, Any]] = []
+        if isinstance(value, FileSegment):
+            files.append(value.value.to_dict())
+        elif isinstance(value, ArrayFileSegment):
+            files.extend([i.to_dict() for i in value.value])
+        elif isinstance(value, File):
+            files.append(value.to_dict())
+        elif isinstance(value, list):
            for item in value:
-                file = self._get_file_var_from_value(item)
+                file = cls._get_file_var_from_value(item)
                if file:
                    files.append(file)
-        elif isinstance(value, dict):
-            file = self._get_file_var_from_value(value)
+        elif isinstance(
+            value,
+            dict,
+        ):
+            file = cls._get_file_var_from_value(value)
            if file:
                files.append(file)

        return files

-    def _get_file_var_from_value(self, value: Union[dict, list]) -> Mapping[str, Any] | None:
+    @classmethod
+    def _get_file_var_from_value(cls, value: Union[dict, list]) -> Mapping[str, Any] | None:
        """
        Get file var from value
        :param value: variable value
@@ -101,6 +101,11 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
            )

        # parse files
+        # TODO(QuantumGhost): Move file parsing logic to the API controller layer
+        # for better separation of concerns.
+        #
+        # For implementation reference, see the `_parse_file` function and
+        # `DraftWorkflowNodeRunApi` class which handle this properly.
        files = args["files"] if args.get("files") else []
        file_extra_config = FileUploadConfigManager.convert(override_model_config_dict or app_model_config.to_dict())
        if file_extra_config:
@@ -196,8 +201,6 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
            try:
                # get message
                message = self._get_message(message_id)
-                if message is None:
-                    raise MessageNotExistsError()

                # chatbot app
                runner = CompletionAppRunner()
@@ -29,6 +29,7 @@ from models.enums import CreatorUserRole
 from models.model import App, AppMode, AppModelConfig, Conversation, EndUser, Message, MessageFile
 from services.errors.app_model_config import AppModelConfigBrokenError
 from services.errors.conversation import ConversationNotExistsError
+from services.errors.message import MessageNotExistsError

 logger = logging.getLogger(__name__)

@@ -251,7 +252,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):

        return introduction or ""

-    def _get_conversation(self, conversation_id: str):
+    def _get_conversation(self, conversation_id: str) -> Conversation:
        """
        Get conversation by conversation id
        :param conversation_id: conversation id
@@ -260,11 +261,11 @@ class MessageBasedAppGenerator(BaseAppGenerator):
        conversation = db.session.query(Conversation).filter(Conversation.id == conversation_id).first()

        if not conversation:
-            raise ConversationNotExistsError()
+            raise ConversationNotExistsError("Conversation not exists")

        return conversation

-    def _get_message(self, message_id: str) -> Optional[Message]:
+    def _get_message(self, message_id: str) -> Message:
        """
        Get message by message id
        :param message_id: message id
@@ -272,4 +273,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
        """
        message = db.session.query(Message).filter(Message.id == message_id).first()

+        if message is None:
+            raise MessageNotExistsError("Message not exists")
+
        return message
@@ -25,13 +25,16 @@ from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
 from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
+from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
 from extensions.ext_database import db
 from factories import file_factory
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, App, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
 from models.enums import WorkflowRunTriggeredFrom
+from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService

 logger = logging.getLogger(__name__)

@@ -94,6 +97,11 @@ class WorkflowAppGenerator(BaseAppGenerator):
        files: Sequence[Mapping[str, Any]] = args.get("files") or []

        # parse files
+        # TODO(QuantumGhost): Move file parsing logic to the API controller layer
+        # for better separation of concerns.
+        #
+        # For implementation reference, see the `_parse_file` function and
+        # `DraftWorkflowNodeRunApi` class which handle this properly.
        file_extra_config = FileUploadConfigManager.convert(workflow.features_dict, is_vision=False)
        system_files = file_factory.build_from_mappings(
            mappings=files,
@@ -186,6 +194,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
        streaming: bool = True,
        workflow_thread_pool_id: Optional[str] = None,
+        variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER,
    ) -> Union[Mapping[str, Any], Generator[str | Mapping[str, Any], None, None]]:
        """
        Generate App response.
@@ -211,6 +220,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
        # new thread with request context and contextvars
        context = contextvars.copy_context()

+        # release database connection, because the following new thread operations may take a long time
+        db.session.close()
+
        worker_thread = threading.Thread(
            target=self._generate_worker,
            kwargs={
@@ -219,11 +231,16 @@ class WorkflowAppGenerator(BaseAppGenerator):
                "queue_manager": queue_manager,
                "context": context,
                "workflow_thread_pool_id": workflow_thread_pool_id,
+                "variable_loader": variable_loader,
            },
        )

        worker_thread.start()

+        draft_var_saver_factory = self._get_draft_var_saver_factory(
+            invoke_from,
+        )
+
        # return response or stream generator
        response = self._handle_response(
            application_generate_entity=application_generate_entity,
@@ -232,6 +249,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            user=user,
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
+            draft_var_saver_factory=draft_var_saver_factory,
            stream=streaming,
        )

@@ -303,6 +321,13 @@ class WorkflowAppGenerator(BaseAppGenerator):
            app_id=application_generate_entity.app_config.app_id,
            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )
+        draft_var_srv = WorkflowDraftVariableService(db.session())
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
+        var_loader = DraftVarLoader(
+            engine=db.engine,
+            app_id=application_generate_entity.app_config.app_id,
+            tenant_id=application_generate_entity.app_config.tenant_id,
+        )

        return self._generate(
            app_model=app_model,
@@ -313,6 +338,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            streaming=streaming,
+            variable_loader=var_loader,
        )

    def single_loop_generate(
@@ -379,7 +405,13 @@ class WorkflowAppGenerator(BaseAppGenerator):
            app_id=application_generate_entity.app_config.app_id,
            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )
-
+        draft_var_srv = WorkflowDraftVariableService(db.session())
+        draft_var_srv.prefill_conversation_variable_default_values(workflow)
+        var_loader = DraftVarLoader(
+            engine=db.engine,
+            app_id=application_generate_entity.app_config.app_id,
+            tenant_id=application_generate_entity.app_config.tenant_id,
+        )
        return self._generate(
            app_model=app_model,
            workflow=workflow,
@@ -389,6 +421,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            streaming=streaming,
+            variable_loader=var_loader,
        )

    def _generate_worker(
@@ -397,6 +430,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
        application_generate_entity: WorkflowAppGenerateEntity,
        queue_manager: AppQueueManager,
        context: contextvars.Context,
+        variable_loader: VariableLoader,
        workflow_thread_pool_id: Optional[str] = None,
    ) -> None:
        """
@@ -415,6 +449,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
                    application_generate_entity=application_generate_entity,
                    queue_manager=queue_manager,
                    workflow_thread_pool_id=workflow_thread_pool_id,
+                    variable_loader=variable_loader,
                )

                runner.run()
@@ -445,6 +480,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
        user: Union[Account, EndUser],
        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        draft_var_saver_factory: DraftVariableSaverFactory,
        stream: bool = False,
    ) -> Union[WorkflowAppBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]]:
        """
@@ -465,6 +501,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            user=user,
            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
+            draft_var_saver_factory=draft_var_saver_factory,
            stream=stream,
        )

@@ -12,6 +12,7 @@ from core.app.entities.app_invoke_entities import (
 from core.workflow.callbacks import WorkflowCallback, WorkflowLoggingCallback
 from core.workflow.entities.variable_pool import VariablePool
 from core.workflow.enums import SystemVariableKey
+from core.workflow.variable_loader import VariableLoader
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from models.enums import UserFrom
@@ -30,6 +31,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
        self,
        application_generate_entity: WorkflowAppGenerateEntity,
        queue_manager: AppQueueManager,
+        variable_loader: VariableLoader,
        workflow_thread_pool_id: Optional[str] = None,
    ) -> None:
        """
@@ -37,10 +39,13 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
        :param queue_manager: application queue manager
        :param workflow_thread_pool_id: workflow thread pool id
        """
+        super().__init__(queue_manager, variable_loader)
        self.application_generate_entity = application_generate_entity
-        self.queue_manager = queue_manager
        self.workflow_thread_pool_id = workflow_thread_pool_id

+    def _get_app_id(self) -> str:
+        return self.application_generate_entity.app_config.app_id
+
    def run(self) -> None:
        """
        Run application
@@ -56,6 +56,7 @@ from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.entities.workflow_execution import WorkflowExecution, WorkflowExecutionStatus, WorkflowType
 from core.workflow.enums import SystemVariableKey
+from core.workflow.repositories.draft_variable_repository import DraftVariableSaverFactory
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
@@ -87,6 +88,7 @@ class WorkflowAppGenerateTaskPipeline:
        stream: bool,
        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
+        draft_var_saver_factory: DraftVariableSaverFactory,
    ) -> None:
        self._base_task_pipeline = BasedGenerateTaskPipeline(
            application_generate_entity=application_generate_entity,
@@ -131,6 +133,8 @@ class WorkflowAppGenerateTaskPipeline:
        self._application_generate_entity = application_generate_entity
        self._workflow_features_dict = workflow.features_dict
        self._workflow_run_id = ""
+        self._invoke_from = queue_manager._invoke_from
+        self._draft_var_saver_factory = draft_var_saver_factory

    def process(self) -> Union[WorkflowAppBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]]:
        """
@@ -322,6 +326,8 @@ class WorkflowAppGenerateTaskPipeline:
                    workflow_node_execution=workflow_node_execution,
                )

+                self._save_output_for_event(event, workflow_node_execution.id)
+
                if node_success_response:
                    yield node_success_response
            elif isinstance(
@@ -339,6 +345,8 @@ class WorkflowAppGenerateTaskPipeline:
                    task_id=self._application_generate_entity.task_id,
                    workflow_node_execution=workflow_node_execution,
                )
+                if isinstance(event, QueueNodeExceptionEvent):
+                    self._save_output_for_event(event, workflow_node_execution.id)

                if node_failed_response:
                    yield node_failed_response
@@ -593,3 +601,15 @@ class WorkflowAppGenerateTaskPipeline:
        )

        return response
+
+    def _save_output_for_event(self, event: QueueNodeSucceededEvent | QueueNodeExceptionEvent, node_execution_id: str):
+        with Session(db.engine) as session, session.begin():
+            saver = self._draft_var_saver_factory(
+                session=session,
+                app_id=self._application_generate_entity.app_config.app_id,
+                node_id=event.node_id,
+                node_type=event.node_type,
+                node_execution_id=node_execution_id,
+                enclosing_node_id=event.in_loop_id or event.in_iteration_id,
+            )
+            saver.save(event.process_data, event.outputs)
@@ -62,6 +62,7 @@ from core.workflow.graph_engine.entities.event import (
 from core.workflow.graph_engine.entities.graph import Graph
 from core.workflow.nodes import NodeType
 from core.workflow.nodes.node_mapping import NODE_TYPE_CLASSES_MAPPING
+from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader, load_into_variable_pool
 from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_database import db
 from models.model import App
@@ -69,8 +70,12 @@ from models.workflow import Workflow


 class WorkflowBasedAppRunner(AppRunner):
-    def __init__(self, queue_manager: AppQueueManager):
+    def __init__(self, queue_manager: AppQueueManager, variable_loader: VariableLoader = DUMMY_VARIABLE_LOADER) -> None:
        self.queue_manager = queue_manager
+        self._variable_loader = variable_loader
+
+    def _get_app_id(self) -> str:
+        raise NotImplementedError("not implemented")

    def _init_graph(self, graph_config: Mapping[str, Any]) -> Graph:
        """
@@ -173,6 +178,13 @@ class WorkflowBasedAppRunner(AppRunner):
        except NotImplementedError:
            variable_mapping = {}

+        load_into_variable_pool(
+            variable_loader=self._variable_loader,
+            variable_pool=variable_pool,
+            variable_mapping=variable_mapping,
+            user_inputs=user_inputs,
+        )
+
        WorkflowEntry.mapping_user_inputs_to_variable_pool(
            variable_mapping=variable_mapping,
            user_inputs=user_inputs,
@@ -262,6 +274,12 @@ class WorkflowBasedAppRunner(AppRunner):
            )
        except NotImplementedError:
            variable_mapping = {}
+        load_into_variable_pool(
+            self._variable_loader,
+            variable_pool=variable_pool,
+            variable_mapping=variable_mapping,
+            user_inputs=user_inputs,
+        )

        WorkflowEntry.mapping_user_inputs_to_variable_pool(
            variable_mapping=variable_mapping,
@@ -376,6 +394,7 @@ class WorkflowBasedAppRunner(AppRunner):
                    in_loop_id=event.in_loop_id,
                )
            )
+
        elif isinstance(event, NodeRunFailedEvent):
            self._publish_event(
                QueueNodeFailedEvent(
@@ -438,6 +457,7 @@ class WorkflowBasedAppRunner(AppRunner):
                    in_loop_id=event.in_loop_id,
                )
            )
+
        elif isinstance(event, NodeInIterationFailedEvent):
            self._publish_event(
                QueueNodeInIterationFailedEvent(
@@ -17,9 +17,24 @@ class InvokeFrom(Enum):
    Invoke From.
    """

+    # SERVICE_API indicates that this invocation is from an API call to Dify app.
+    #
+    # Description of service api in Dify docs:
+    # https://docs.dify.ai/en/guides/application-publishing/developing-with-apis
    SERVICE_API = "service-api"
+
+    # WEB_APP indicates that this invocation is from
+    # the web app of the workflow (or chatflow).
+    #
+    # Description of web app in Dify docs:
+    # https://docs.dify.ai/en/guides/application-publishing/launch-your-webapp-quickly/README
    WEB_APP = "web-app"
+
+    # EXPLORE indicates that this invocation is from
+    # the workflow (or chatflow) explore page.
    EXPLORE = "explore"
+    # DEBUGGER indicates that this invocation is from
+    # the workflow (or chatflow) edit page.
    DEBUGGER = "debugger"

    @classmethod
@@ -19,6 +19,7 @@ from core.app.entities.task_entities import (
 from core.errors.error import QuotaExceededError
 from core.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeError
 from core.moderation.output_moderation import ModerationRule, OutputModeration
+from models.enums import MessageStatus
 from models.model import Message

 logger = logging.getLogger(__name__)
@@ -62,7 +63,7 @@ class BasedGenerateTaskPipeline:
            return err

        err_desc = self._error_to_desc(err)
-        message.status = "error"
+        message.status = MessageStatus.ERROR
        message.error = err_desc
        return err

@@ -395,6 +395,7 @@ class EasyUIBasedGenerateTaskPipeline(BasedGenerateTaskPipeline):
        message.provider_response_latency = time.perf_counter() - self._start_at
        message.total_price = usage.total_price
        message.currency = usage.currency
+        self._task_state.llm_result.usage.latency = message.provider_response_latency
        message.message_metadata = self._task_state.metadata.model_dump_json()

        if trace_manager:
@@ -15,7 +15,15 @@ class CommonParameterType(StrEnum):
    MODEL_SELECTOR = "model-selector"
    TOOLS_SELECTOR = "array[tools]"

+    # Dynamic select parameter
+    # Once you are not sure about the available options until authorization is done
+    # eg: Select a Slack channel from a Slack workspace
+    DYNAMIC_SELECT = "dynamic-select"
+
    # TOOL_SELECTOR = "tool-selector"
+    # MCP object and array type parameters
+    ARRAY = "array"
+    OBJECT = "object"


 class AppSelectorScope(StrEnum):
@@ -1 +1,11 @@
+from typing import Any
+
+# TODO(QuantumGhost): Refactor variable type identification. Instead of directly
+# comparing `dify_model_identity` with constants throughout the codebase, extract
+# this logic into a dedicated function. This would encapsulate the implementation
+# details of how different variable types are identified.
 FILE_MODEL_IDENTITY = "__dify__file__"
+
+
+def maybe_file_object(o: Any) -> bool:
+    return isinstance(o, dict) and o.get("dify_model_identity") == FILE_MODEL_IDENTITY
@@ -51,7 +51,7 @@ class File(BaseModel):
    # It should be set to `ToolFile.id` when `transfer_method` is `tool_file`.
    related_id: Optional[str] = None
    filename: Optional[str] = None
-    extension: Optional[str] = Field(default=None, description="File extension, should contains dot")
+    extension: Optional[str] = Field(default=None, description="File extension, should contain dot")
    mime_type: Optional[str] = None
    size: int = -1

@@ -1,67 +0,0 @@
-import base64
-import logging
-import time
-from typing import Optional
-
-from configs import dify_config
-from constants import IMAGE_EXTENSIONS
-from core.helper.url_signer import UrlSigner
-from extensions.ext_storage import storage
-
-
-class UploadFileParser:
-    @classmethod
-    def get_image_data(cls, upload_file, force_url: bool = False) -> Optional[str]:
-        if not upload_file:
-            return None
-
-        if upload_file.extension not in IMAGE_EXTENSIONS:
-            return None
-
-        if dify_config.MULTIMODAL_SEND_FORMAT == "url" or force_url:
-            return cls.get_signed_temp_image_url(upload_file.id)
-        else:
-            # get image file base64
-            try:
-                data = storage.load(upload_file.key)
-            except FileNotFoundError:
-                logging.exception(f"File not found: {upload_file.key}")
-                return None
-
-            encoded_string = base64.b64encode(data).decode("utf-8")
-            return f"data:{upload_file.mime_type};base64,{encoded_string}"
-
-    @classmethod
-    def get_signed_temp_image_url(cls, upload_file_id) -> str:
-        """
-        get signed url from upload file
-
-        :param upload_file_id: the id of UploadFile object
-        :return:
-        """
-        base_url = dify_config.FILES_URL
-        image_preview_url = f"{base_url}/files/{upload_file_id}/image-preview"
-
-        return UrlSigner.get_signed_url(url=image_preview_url, sign_key=upload_file_id, prefix="image-preview")
-
-    @classmethod
-    def verify_image_file_signature(cls, upload_file_id: str, timestamp: str, nonce: str, sign: str) -> bool:
-        """
-        verify signature
-
-        :param upload_file_id: file id
-        :param timestamp: timestamp
-        :param nonce: nonce
-        :param sign: signature
-        :return:
-        """
-        result = UrlSigner.verify(
-            sign_key=upload_file_id, timestamp=timestamp, nonce=nonce, sign=sign, prefix="image-preview"
-        )
-
-        # verify signature
-        if not result:
-            return False
-
-        current_time = int(time.time())
-        return current_time - int(timestamp) <= dify_config.FILES_ACCESS_TIMEOUT
@@ -28,7 +28,7 @@ class TemplateTransformer(ABC):
    def extract_result_str_from_response(cls, response: str):
        result = re.search(rf"{cls._result_tag}(.*){cls._result_tag}", response, re.DOTALL)
        if not result:
-            raise ValueError("Failed to parse result")
+            raise ValueError(f"Failed to parse result: no result tag found in response. Response: {response[:200]}...")
        return result.group(1)

    @classmethod
@@ -38,16 +38,53 @@ class TemplateTransformer(ABC):
        :param response: response
        :return:
        """
+
        try:
-            result = json.loads(cls.extract_result_str_from_response(response))
-        except json.JSONDecodeError:
-            raise ValueError("failed to parse response")
+            result_str = cls.extract_result_str_from_response(response)
+            result = json.loads(result_str)
+        except json.JSONDecodeError as e:
+            raise ValueError(f"Failed to parse JSON response: {str(e)}. Response content: {result_str[:200]}...")
+        except ValueError as e:
+            # Re-raise ValueError from extract_result_str_from_response
+            raise e
+        except Exception as e:
+            raise ValueError(f"Unexpected error during response transformation: {str(e)}")
+
+        # Check if the result contains an error
+        if isinstance(result, dict) and "error" in result:
+            raise ValueError(f"JavaScript execution error: {result['error']}")
+
        if not isinstance(result, dict):
-            raise ValueError("result must be a dict")
+            raise ValueError(f"Result must be a dict, got {type(result).__name__}")
        if not all(isinstance(k, str) for k in result):
-            raise ValueError("result keys must be strings")
+            raise ValueError("Result keys must be strings")
+
+        # Post-process the result to convert scientific notation strings back to numbers
+        result = cls._post_process_result(result)
        return result

+    @classmethod
+    def _post_process_result(cls, result: dict[Any, Any]) -> dict[Any, Any]:
+        """
+        Post-process the result to convert scientific notation strings back to numbers
+        """
+
+        def convert_scientific_notation(value):
+            if isinstance(value, str):
+                # Check if the string looks like scientific notation
+                if re.match(r"^-?\d+\.?\d*e[+-]\d+$", value, re.IGNORECASE):
+                    try:
+                        return float(value)
+                    except ValueError:
+                        pass
+            elif isinstance(value, dict):
+                return {k: convert_scientific_notation(v) for k, v in value.items()}
+            elif isinstance(value, list):
+                return [convert_scientific_notation(v) for v in value]
+            return value
+
+        return convert_scientific_notation(result)  # type: ignore[no-any-return]
+
    @classmethod
    @abstractmethod
    def get_runner_script(cls) -> str:
@@ -1,22 +0,0 @@
-from collections import OrderedDict
-from typing import Any
-
-
-class LRUCache:
-    def __init__(self, capacity: int):
-        self.cache: OrderedDict[Any, Any] = OrderedDict()
-        self.capacity = capacity
-
-    def get(self, key: Any) -> Any:
-        if key not in self.cache:
-            return None
-        else:
-            self.cache.move_to_end(key)  # move the key to the end of the OrderedDict
-            return self.cache[key]
-
-    def put(self, key: Any, value: Any) -> None:
-        if key in self.cache:
-            self.cache.move_to_end(key)
-        self.cache[key] = value
-        if len(self.cache) > self.capacity:
-            self.cache.popitem(last=False)  # pop the first item
@@ -25,9 +25,29 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
    url = str(marketplace_api_url / "api/v1/plugins/batch")
    response = requests.post(url, json={"plugin_ids": plugin_ids})
    response.raise_for_status()
+
    return [MarketplacePluginDeclaration(**plugin) for plugin in response.json()["data"]["plugins"]]


+def batch_fetch_plugin_manifests_ignore_deserialization_error(
+    plugin_ids: list[str],
+) -> Sequence[MarketplacePluginDeclaration]:
+    if len(plugin_ids) == 0:
+        return []
+
+    url = str(marketplace_api_url / "api/v1/plugins/batch")
+    response = requests.post(url, json={"plugin_ids": plugin_ids})
+    response.raise_for_status()
+    result: list[MarketplacePluginDeclaration] = []
+    for plugin in response.json()["data"]["plugins"]:
+        try:
+            result.append(MarketplacePluginDeclaration(**plugin))
+        except Exception as e:
+            pass
+
+    return result
+
+
 def record_install_plugin_event(plugin_unique_identifier: str):
    url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
    response = requests.post(url, json={"unique_identifier": plugin_unique_identifier})
@@ -317,9 +317,10 @@ class IndexingRunner:
                image_upload_file_ids = get_image_upload_file_ids(document.page_content)
                for upload_file_id in image_upload_file_ids:
                    image_file = db.session.query(UploadFile).filter(UploadFile.id == upload_file_id).first()
+                    if image_file is None:
+                        continue
                    try:
-                        if image_file:
-                            storage.delete(image_file.key)
+                        storage.delete(image_file.key)
                    except Exception:
                        logging.exception(
                            "Delete image_files failed while indexing_estimate, \
@@ -534,7 +535,7 @@ class IndexingRunner:
        # chunk nodes by chunk size
        indexing_start_at = time.perf_counter()
        tokens = 0
-        if dataset_document.doc_form != IndexType.PARENT_CHILD_INDEX:
+        if dataset_document.doc_form != IndexType.PARENT_CHILD_INDEX and dataset.indexing_technique == "economy":
            # create keyword index
            create_keyword_thread = threading.Thread(
                target=self._process_keyword_index,
@@ -572,7 +573,7 @@ class IndexingRunner:

                for future in futures:
                    tokens += future.result()
-        if dataset_document.doc_form != IndexType.PARENT_CHILD_INDEX:
+        if dataset_document.doc_form != IndexType.PARENT_CHILD_INDEX and dataset.indexing_technique == "economy":
            create_keyword_thread.join()
        indexing_end_at = time.perf_counter()

@@ -0,0 +1,380 @@
+import json
+from collections.abc import Generator, Mapping, Sequence
+from copy import deepcopy
+from enum import StrEnum
+from typing import Any, Literal, Optional, cast, overload
+
+import json_repair
+from pydantic import TypeAdapter, ValidationError
+
+from core.llm_generator.output_parser.errors import OutputParserError
+from core.llm_generator.prompts import STRUCTURED_OUTPUT_PROMPT
+from core.model_manager import ModelInstance
+from core.model_runtime.callbacks.base_callback import Callback
+from core.model_runtime.entities.llm_entities import (
+    LLMResult,
+    LLMResultChunk,
+    LLMResultChunkDelta,
+    LLMResultChunkWithStructuredOutput,
+    LLMResultWithStructuredOutput,
+)
+from core.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    PromptMessage,
+    PromptMessageTool,
+    SystemPromptMessage,
+    TextPromptMessageContent,
+)
+from core.model_runtime.entities.model_entities import AIModelEntity, ParameterRule
+
+
+class ResponseFormat(StrEnum):
+    """Constants for model response formats"""
+
+    JSON_SCHEMA = "json_schema"  # model's structured output mode. some model like gemini, gpt-4o,  support this mode.
+    JSON = "JSON"  # model's json mode. some model like claude support this mode.
+    JSON_OBJECT = "json_object"  # json mode's another alias. some model like deepseek-chat, qwen use this alias.
+
+
+class SpecialModelType(StrEnum):
+    """Constants for identifying model types"""
+
+    GEMINI = "gemini"
+    OLLAMA = "ollama"
+
+
+@overload
+def invoke_llm_with_structured_output(
+    provider: str,
+    model_schema: AIModelEntity,
+    model_instance: ModelInstance,
+    prompt_messages: Sequence[PromptMessage],
+    json_schema: Mapping[str, Any],
+    model_parameters: Optional[Mapping] = None,
+    tools: Sequence[PromptMessageTool] | None = None,
+    stop: Optional[list[str]] = None,
+    stream: Literal[True] = True,
+    user: Optional[str] = None,
+    callbacks: Optional[list[Callback]] = None,
+) -> Generator[LLMResultChunkWithStructuredOutput, None, None]: ...
+
+
+@overload
+def invoke_llm_with_structured_output(
+    provider: str,
+    model_schema: AIModelEntity,
+    model_instance: ModelInstance,
+    prompt_messages: Sequence[PromptMessage],
+    json_schema: Mapping[str, Any],
+    model_parameters: Optional[Mapping] = None,
+    tools: Sequence[PromptMessageTool] | None = None,
+    stop: Optional[list[str]] = None,
+    stream: Literal[False] = False,
+    user: Optional[str] = None,
+    callbacks: Optional[list[Callback]] = None,
+) -> LLMResultWithStructuredOutput: ...
+
+
+@overload
+def invoke_llm_with_structured_output(
+    provider: str,
+    model_schema: AIModelEntity,
+    model_instance: ModelInstance,
+    prompt_messages: Sequence[PromptMessage],
+    json_schema: Mapping[str, Any],
+    model_parameters: Optional[Mapping] = None,
+    tools: Sequence[PromptMessageTool] | None = None,
+    stop: Optional[list[str]] = None,
+    stream: bool = True,
+    user: Optional[str] = None,
+    callbacks: Optional[list[Callback]] = None,
+) -> LLMResultWithStructuredOutput | Generator[LLMResultChunkWithStructuredOutput, None, None]: ...
+
+
+def invoke_llm_with_structured_output(
+    provider: str,
+    model_schema: AIModelEntity,
+    model_instance: ModelInstance,
+    prompt_messages: Sequence[PromptMessage],
+    json_schema: Mapping[str, Any],
+    model_parameters: Optional[Mapping] = None,
+    tools: Sequence[PromptMessageTool] | None = None,
+    stop: Optional[list[str]] = None,
+    stream: bool = True,
+    user: Optional[str] = None,
+    callbacks: Optional[list[Callback]] = None,
+) -> LLMResultWithStructuredOutput | Generator[LLMResultChunkWithStructuredOutput, None, None]:
+    """
+    Invoke large language model with structured output
+    1. This method invokes model_instance.invoke_llm with json_schema
+    2. Try to parse the result as structured output
+
+    :param prompt_messages: prompt messages
+    :param json_schema: json schema
+    :param model_parameters: model parameters
+    :param tools: tools for tool calling
+    :param stop: stop words
+    :param stream: is stream response
+    :param user: unique user id
+    :param callbacks: callbacks
+    :return: full response or stream response chunk generator result
+    """
+
+    # handle native json schema
+    model_parameters_with_json_schema: dict[str, Any] = {
+        **(model_parameters or {}),
+    }
+
+    if model_schema.support_structure_output:
+        model_parameters = _handle_native_json_schema(
+            provider, model_schema, json_schema, model_parameters_with_json_schema, model_schema.parameter_rules
+        )
+    else:
+        # Set appropriate response format based on model capabilities
+        _set_response_format(model_parameters_with_json_schema, model_schema.parameter_rules)
+
+        # handle prompt based schema
+        prompt_messages = _handle_prompt_based_schema(
+            prompt_messages=prompt_messages,
+            structured_output_schema=json_schema,
+        )
+
+    llm_result = model_instance.invoke_llm(
+        prompt_messages=list(prompt_messages),
+        model_parameters=model_parameters_with_json_schema,
+        tools=tools,
+        stop=stop,
+        stream=stream,
+        user=user,
+        callbacks=callbacks,
+    )
+
+    if isinstance(llm_result, LLMResult):
+        if not isinstance(llm_result.message.content, str):
+            raise OutputParserError(
+                f"Failed to parse structured output, LLM result is not a string: {llm_result.message.content}"
+            )
+
+        return LLMResultWithStructuredOutput(
+            structured_output=_parse_structured_output(llm_result.message.content),
+            model=llm_result.model,
+            message=llm_result.message,
+            usage=llm_result.usage,
+            system_fingerprint=llm_result.system_fingerprint,
+            prompt_messages=llm_result.prompt_messages,
+        )
+    else:
+
+        def generator() -> Generator[LLMResultChunkWithStructuredOutput, None, None]:
+            result_text: str = ""
+            prompt_messages: Sequence[PromptMessage] = []
+            system_fingerprint: Optional[str] = None
+            for event in llm_result:
+                if isinstance(event, LLMResultChunk):
+                    prompt_messages = event.prompt_messages
+                    system_fingerprint = event.system_fingerprint
+
+                    if isinstance(event.delta.message.content, str):
+                        result_text += event.delta.message.content
+                    elif isinstance(event.delta.message.content, list):
+                        for item in event.delta.message.content:
+                            if isinstance(item, TextPromptMessageContent):
+                                result_text += item.data
+
+                yield LLMResultChunkWithStructuredOutput(
+                    model=model_schema.model,
+                    prompt_messages=prompt_messages,
+                    system_fingerprint=system_fingerprint,
+                    delta=event.delta,
+                )
+
+            yield LLMResultChunkWithStructuredOutput(
+                structured_output=_parse_structured_output(result_text),
+                model=model_schema.model,
+                prompt_messages=prompt_messages,
+                system_fingerprint=system_fingerprint,
+                delta=LLMResultChunkDelta(
+                    index=0,
+                    message=AssistantPromptMessage(content=""),
+                    usage=None,
+                    finish_reason=None,
+                ),
+            )
+
+        return generator()
+
+
+def _handle_native_json_schema(
+    provider: str,
+    model_schema: AIModelEntity,
+    structured_output_schema: Mapping,
+    model_parameters: dict,
+    rules: list[ParameterRule],
+) -> dict:
+    """
+    Handle structured output for models with native JSON schema support.
+
+    :param model_parameters: Model parameters to update
+    :param rules: Model parameter rules
+    :return: Updated model parameters with JSON schema configuration
+    """
+    # Process schema according to model requirements
+    schema_json = _prepare_schema_for_model(provider, model_schema, structured_output_schema)
+
+    # Set JSON schema in parameters
+    model_parameters["json_schema"] = json.dumps(schema_json, ensure_ascii=False)
+
+    # Set appropriate response format if required by the model
+    for rule in rules:
+        if rule.name == "response_format" and ResponseFormat.JSON_SCHEMA.value in rule.options:
+            model_parameters["response_format"] = ResponseFormat.JSON_SCHEMA.value
+
+    return model_parameters
+
+
+def _set_response_format(model_parameters: dict, rules: list) -> None:
+    """
+    Set the appropriate response format parameter based on model rules.
+
+    :param model_parameters: Model parameters to update
+    :param rules: Model parameter rules
+    """
+    for rule in rules:
+        if rule.name == "response_format":
+            if ResponseFormat.JSON.value in rule.options:
+                model_parameters["response_format"] = ResponseFormat.JSON.value
+            elif ResponseFormat.JSON_OBJECT.value in rule.options:
+                model_parameters["response_format"] = ResponseFormat.JSON_OBJECT.value
+
+
+def _handle_prompt_based_schema(
+    prompt_messages: Sequence[PromptMessage], structured_output_schema: Mapping
+) -> list[PromptMessage]:
+    """
+    Handle structured output for models without native JSON schema support.
+    This function modifies the prompt messages to include schema-based output requirements.
+
+    Args:
+        prompt_messages: Original sequence of prompt messages
+
+    Returns:
+        list[PromptMessage]: Updated prompt messages with structured output requirements
+    """
+    # Convert schema to string format
+    schema_str = json.dumps(structured_output_schema, ensure_ascii=False)
+
+    # Find existing system prompt with schema placeholder
+    system_prompt = next(
+        (prompt for prompt in prompt_messages if isinstance(prompt, SystemPromptMessage)),
+        None,
+    )
+    structured_output_prompt = STRUCTURED_OUTPUT_PROMPT.replace("{{schema}}", schema_str)
+    # Prepare system prompt content
+    system_prompt_content = (
+        structured_output_prompt + "\n\n" + system_prompt.content
+        if system_prompt and isinstance(system_prompt.content, str)
+        else structured_output_prompt
+    )
+    system_prompt = SystemPromptMessage(content=system_prompt_content)
+
+    # Extract content from the last user message
+
+    filtered_prompts = [prompt for prompt in prompt_messages if not isinstance(prompt, SystemPromptMessage)]
+    updated_prompt = [system_prompt] + filtered_prompts
+
+    return updated_prompt
+
+
+def _parse_structured_output(result_text: str) -> Mapping[str, Any]:
+    structured_output: Mapping[str, Any] = {}
+    parsed: Mapping[str, Any] = {}
+    try:
+        parsed = TypeAdapter(Mapping).validate_json(result_text)
+        if not isinstance(parsed, dict):
+            raise OutputParserError(f"Failed to parse structured output: {result_text}")
+        structured_output = parsed
+    except ValidationError:
+        # if the result_text is not a valid json, try to repair it
+        temp_parsed = json_repair.loads(result_text)
+        if not isinstance(temp_parsed, dict):
+            # handle reasoning model like deepseek-r1 got '<think>\n\n</think>\n' prefix
+            if isinstance(temp_parsed, list):
+                temp_parsed = next((item for item in temp_parsed if isinstance(item, dict)), {})
+            else:
+                raise OutputParserError(f"Failed to parse structured output: {result_text}")
+        structured_output = cast(dict, temp_parsed)
+    return structured_output
+
+
+def _prepare_schema_for_model(provider: str, model_schema: AIModelEntity, schema: Mapping) -> dict:
+    """
+    Prepare JSON schema based on model requirements.
+
+    Different models have different requirements for JSON schema formatting.
+    This function handles these differences.
+
+    :param schema: The original JSON schema
+    :return: Processed schema compatible with the current model
+    """
+
+    # Deep copy to avoid modifying the original schema
+    processed_schema = dict(deepcopy(schema))
+
+    # Convert boolean types to string types (common requirement)
+    convert_boolean_to_string(processed_schema)
+
+    # Apply model-specific transformations
+    if SpecialModelType.GEMINI in model_schema.model:
+        remove_additional_properties(processed_schema)
+        return processed_schema
+    elif SpecialModelType.OLLAMA in provider:
+        return processed_schema
+    else:
+        # Default format with name field
+        return {"schema": processed_schema, "name": "llm_response"}
+
+
+def remove_additional_properties(schema: dict) -> None:
+    """
+    Remove additionalProperties fields from JSON schema.
+    Used for models like Gemini that don't support this property.
+
+    :param schema: JSON schema to modify in-place
+    """
+    if not isinstance(schema, dict):
+        return
+
+    # Remove additionalProperties at current level
+    schema.pop("additionalProperties", None)
+
+    # Process nested structures recursively
+    for value in schema.values():
+        if isinstance(value, dict):
+            remove_additional_properties(value)
+        elif isinstance(value, list):
+            for item in value:
+                if isinstance(item, dict):
+                    remove_additional_properties(item)
+
+
+def convert_boolean_to_string(schema: dict) -> None:
+    """
+    Convert boolean type specifications to string in JSON schema.
+
+    :param schema: JSON schema to modify in-place
+    """
+    if not isinstance(schema, dict):
+        return
+
+    # Check for boolean type at current level
+    if schema.get("type") == "boolean":
+        schema["type"] = "string"
+
+    # Process nested dictionaries and lists recursively
+    for value in schema.values():
+        if isinstance(value, dict):
+            convert_boolean_to_string(value)
+        elif isinstance(value, list):
+            for item in value:
+                if isinstance(item, dict):
+                    convert_boolean_to_string(item)
@@ -291,3 +291,21 @@ Your task is to convert simple user descriptions into properly formatted JSON Sc

 Now, generate a JSON Schema based on my description
 """  # noqa: E501
+
+STRUCTURED_OUTPUT_PROMPT = """You’re a helpful AI assistant. You could answer questions and output in JSON format.
+constraints:
+    - You must output in JSON format.
+    - Do not output boolean value, use string type instead.
+    - Do not output integer or float value, use number type instead.
+eg:
+    Here is the JSON schema:
+    {"additionalProperties": false, "properties": {"age": {"type": "number"}, "name": {"type": "string"}}, "required": ["name", "age"], "type": "object"}
+
+    Here is the user's question:
+    My name is John Doe and I am 30 years old.
+
+    output:
+    {"name": "John Doe", "age": 30}
+Here is the JSON schema:
+{{schema}}
+"""  # noqa: E501
@@ -0,0 +1,342 @@
+import base64
+import hashlib
+import json
+import os
+import secrets
+import urllib.parse
+from typing import Optional
+from urllib.parse import urljoin
+
+import requests
+from pydantic import BaseModel, ValidationError
+
+from core.mcp.auth.auth_provider import OAuthClientProvider
+from core.mcp.types import (
+    OAuthClientInformation,
+    OAuthClientInformationFull,
+    OAuthClientMetadata,
+    OAuthMetadata,
+    OAuthTokens,
+)
+from extensions.ext_redis import redis_client
+
+LATEST_PROTOCOL_VERSION = "1.0"
+OAUTH_STATE_EXPIRY_SECONDS = 5 * 60  # 5 minutes expiry
+OAUTH_STATE_REDIS_KEY_PREFIX = "oauth_state:"
+
+
+class OAuthCallbackState(BaseModel):
+    provider_id: str
+    tenant_id: str
+    server_url: str
+    metadata: OAuthMetadata | None = None
+    client_information: OAuthClientInformation
+    code_verifier: str
+    redirect_uri: str
+
+
+def generate_pkce_challenge() -> tuple[str, str]:
+    """Generate PKCE challenge and verifier."""
+    code_verifier = base64.urlsafe_b64encode(os.urandom(40)).decode("utf-8")
+    code_verifier = code_verifier.replace("=", "").replace("+", "-").replace("/", "_")
+
+    code_challenge_hash = hashlib.sha256(code_verifier.encode("utf-8")).digest()
+    code_challenge = base64.urlsafe_b64encode(code_challenge_hash).decode("utf-8")
+    code_challenge = code_challenge.replace("=", "").replace("+", "-").replace("/", "_")
+
+    return code_verifier, code_challenge
+
+
+def _create_secure_redis_state(state_data: OAuthCallbackState) -> str:
+    """Create a secure state parameter by storing state data in Redis and returning a random state key."""
+    # Generate a secure random state key
+    state_key = secrets.token_urlsafe(32)
+
+    # Store the state data in Redis with expiration
+    redis_key = f"{OAUTH_STATE_REDIS_KEY_PREFIX}{state_key}"
+    redis_client.setex(redis_key, OAUTH_STATE_EXPIRY_SECONDS, state_data.model_dump_json())
+
+    return state_key
+
+
+def _retrieve_redis_state(state_key: str) -> OAuthCallbackState:
+    """Retrieve and decode OAuth state data from Redis using the state key, then delete it."""
+    redis_key = f"{OAUTH_STATE_REDIS_KEY_PREFIX}{state_key}"
+
+    # Get state data from Redis
+    state_data = redis_client.get(redis_key)
+
+    if not state_data:
+        raise ValueError("State parameter has expired or does not exist")
+
+    # Delete the state data from Redis immediately after retrieval to prevent reuse
+    redis_client.delete(redis_key)
+
+    try:
+        # Parse and validate the state data
+        oauth_state = OAuthCallbackState.model_validate_json(state_data)
+
+        return oauth_state
+    except ValidationError as e:
+        raise ValueError(f"Invalid state parameter: {str(e)}")
+
+
+def handle_callback(state_key: str, authorization_code: str) -> OAuthCallbackState:
+    """Handle the callback from the OAuth provider."""
+    # Retrieve state data from Redis (state is automatically deleted after retrieval)
+    full_state_data = _retrieve_redis_state(state_key)
+
+    tokens = exchange_authorization(
+        full_state_data.server_url,
+        full_state_data.metadata,
+        full_state_data.client_information,
+        authorization_code,
+        full_state_data.code_verifier,
+        full_state_data.redirect_uri,
+    )
+    provider = OAuthClientProvider(full_state_data.provider_id, full_state_data.tenant_id, for_list=True)
+    provider.save_tokens(tokens)
+    return full_state_data
+
+
+def discover_oauth_metadata(server_url: str, protocol_version: Optional[str] = None) -> Optional[OAuthMetadata]:
+    """Looks up RFC 8414 OAuth 2.0 Authorization Server Metadata."""
+    url = urljoin(server_url, "/.well-known/oauth-authorization-server")
+
+    try:
+        headers = {"MCP-Protocol-Version": protocol_version or LATEST_PROTOCOL_VERSION}
+        response = requests.get(url, headers=headers)
+        if response.status_code == 404:
+            return None
+        if not response.ok:
+            raise ValueError(f"HTTP {response.status_code} trying to load well-known OAuth metadata")
+        return OAuthMetadata.model_validate(response.json())
+    except requests.RequestException as e:
+        if isinstance(e, requests.ConnectionError):
+            response = requests.get(url)
+            if response.status_code == 404:
+                return None
+            if not response.ok:
+                raise ValueError(f"HTTP {response.status_code} trying to load well-known OAuth metadata")
+            return OAuthMetadata.model_validate(response.json())
+        raise
+
+
+def start_authorization(
+    server_url: str,
+    metadata: Optional[OAuthMetadata],
+    client_information: OAuthClientInformation,
+    redirect_url: str,
+    provider_id: str,
+    tenant_id: str,
+) -> tuple[str, str]:
+    """Begins the authorization flow with secure Redis state storage."""
+    response_type = "code"
+    code_challenge_method = "S256"
+
+    if metadata:
+        authorization_url = metadata.authorization_endpoint
+        if response_type not in metadata.response_types_supported:
+            raise ValueError(f"Incompatible auth server: does not support response type {response_type}")
+        if (
+            not metadata.code_challenge_methods_supported
+            or code_challenge_method not in metadata.code_challenge_methods_supported
+        ):
+            raise ValueError(
+                f"Incompatible auth server: does not support code challenge method {code_challenge_method}"
+            )
+    else:
+        authorization_url = urljoin(server_url, "/authorize")
+
+    code_verifier, code_challenge = generate_pkce_challenge()
+
+    # Prepare state data with all necessary information
+    state_data = OAuthCallbackState(
+        provider_id=provider_id,
+        tenant_id=tenant_id,
+        server_url=server_url,
+        metadata=metadata,
+        client_information=client_information,
+        code_verifier=code_verifier,
+        redirect_uri=redirect_url,
+    )
+
+    # Store state data in Redis and generate secure state key
+    state_key = _create_secure_redis_state(state_data)
+
+    params = {
+        "response_type": response_type,
+        "client_id": client_information.client_id,
+        "code_challenge": code_challenge,
+        "code_challenge_method": code_challenge_method,
+        "redirect_uri": redirect_url,
+        "state": state_key,
+    }
+
+    authorization_url = f"{authorization_url}?{urllib.parse.urlencode(params)}"
+    return authorization_url, code_verifier
+
+
+def exchange_authorization(
+    server_url: str,
+    metadata: Optional[OAuthMetadata],
+    client_information: OAuthClientInformation,
+    authorization_code: str,
+    code_verifier: str,
+    redirect_uri: str,
+) -> OAuthTokens:
+    """Exchanges an authorization code for an access token."""
+    grant_type = "authorization_code"
+
+    if metadata:
+        token_url = metadata.token_endpoint
+        if metadata.grant_types_supported and grant_type not in metadata.grant_types_supported:
+            raise ValueError(f"Incompatible auth server: does not support grant type {grant_type}")
+    else:
+        token_url = urljoin(server_url, "/token")
+
+    params = {
+        "grant_type": grant_type,
+        "client_id": client_information.client_id,
+        "code": authorization_code,
+        "code_verifier": code_verifier,
+        "redirect_uri": redirect_uri,
+    }
+
+    if client_information.client_secret:
+        params["client_secret"] = client_information.client_secret
+
+    response = requests.post(token_url, data=params)
+    if not response.ok:
+        raise ValueError(f"Token exchange failed: HTTP {response.status_code}")
+    return OAuthTokens.model_validate(response.json())
+
+
+def refresh_authorization(
+    server_url: str,
+    metadata: Optional[OAuthMetadata],
+    client_information: OAuthClientInformation,
+    refresh_token: str,
+) -> OAuthTokens:
+    """Exchange a refresh token for an updated access token."""
+    grant_type = "refresh_token"
+
+    if metadata:
+        token_url = metadata.token_endpoint
+        if metadata.grant_types_supported and grant_type not in metadata.grant_types_supported:
+            raise ValueError(f"Incompatible auth server: does not support grant type {grant_type}")
+    else:
+        token_url = urljoin(server_url, "/token")
+
+    params = {
+        "grant_type": grant_type,
+        "client_id": client_information.client_id,
+        "refresh_token": refresh_token,
+    }
+
+    if client_information.client_secret:
+        params["client_secret"] = client_information.client_secret
+
+    response = requests.post(token_url, data=params)
+    if not response.ok:
+        raise ValueError(f"Token refresh failed: HTTP {response.status_code}")
+    return OAuthTokens.parse_obj(response.json())
+
+
+def register_client(
+    server_url: str,
+    metadata: Optional[OAuthMetadata],
+    client_metadata: OAuthClientMetadata,
+) -> OAuthClientInformationFull:
+    """Performs OAuth 2.0 Dynamic Client Registration."""
+    if metadata:
+        if not metadata.registration_endpoint:
+            raise ValueError("Incompatible auth server: does not support dynamic client registration")
+        registration_url = metadata.registration_endpoint
+    else:
+        registration_url = urljoin(server_url, "/register")
+
+    response = requests.post(
+        registration_url,
+        json=client_metadata.model_dump(),
+        headers={"Content-Type": "application/json"},
+    )
+    if not response.ok:
+        response.raise_for_status()
+    return OAuthClientInformationFull.model_validate(response.json())
+
+
+def auth(
+    provider: OAuthClientProvider,
+    server_url: str,
+    authorization_code: Optional[str] = None,
+    state_param: Optional[str] = None,
+    for_list: bool = False,
+) -> dict[str, str]:
+    """Orchestrates the full auth flow with a server using secure Redis state storage."""
+    metadata = discover_oauth_metadata(server_url)
+
+    # Handle client registration if needed
+    client_information = provider.client_information()
+    if not client_information:
+        if authorization_code is not None:
+            raise ValueError("Existing OAuth client information is required when exchanging an authorization code")
+        try:
+            full_information = register_client(server_url, metadata, provider.client_metadata)
+        except requests.RequestException as e:
+            raise ValueError(f"Could not register OAuth client: {e}")
+        provider.save_client_information(full_information)
+        client_information = full_information
+
+    # Exchange authorization code for tokens
+    if authorization_code is not None:
+        if not state_param:
+            raise ValueError("State parameter is required when exchanging authorization code")
+
+        try:
+            # Retrieve state data from Redis using state key
+            full_state_data = _retrieve_redis_state(state_param)
+
+            code_verifier = full_state_data.code_verifier
+            redirect_uri = full_state_data.redirect_uri
+
+            if not code_verifier or not redirect_uri:
+                raise ValueError("Missing code_verifier or redirect_uri in state data")
+
+        except (json.JSONDecodeError, ValueError) as e:
+            raise ValueError(f"Invalid state parameter: {e}")
+
+        tokens = exchange_authorization(
+            server_url,
+            metadata,
+            client_information,
+            authorization_code,
+            code_verifier,
+            redirect_uri,
+        )
+        provider.save_tokens(tokens)
+        return {"result": "success"}
+
+    provider_tokens = provider.tokens()
+
+    # Handle token refresh or new authorization
+    if provider_tokens and provider_tokens.refresh_token:
+        try:
+            new_tokens = refresh_authorization(server_url, metadata, client_information, provider_tokens.refresh_token)
+            provider.save_tokens(new_tokens)
+            return {"result": "success"}
+        except Exception as e:
+            raise ValueError(f"Could not refresh OAuth tokens: {e}")
+
+    # Start new authorization flow
+    authorization_url, code_verifier = start_authorization(
+        server_url,
+        metadata,
+        client_information,
+        provider.redirect_url,
+        provider.mcp_provider.id,
+        provider.mcp_provider.tenant_id,
+    )
+
+    provider.save_code_verifier(code_verifier)
+    return {"authorization_url": authorization_url}
@@ -0,0 +1,81 @@
+from typing import Optional
+
+from configs import dify_config
+from core.mcp.types import (
+    OAuthClientInformation,
+    OAuthClientInformationFull,
+    OAuthClientMetadata,
+    OAuthTokens,
+)
+from models.tools import MCPToolProvider
+from services.tools.mcp_tools_mange_service import MCPToolManageService
+
+LATEST_PROTOCOL_VERSION = "1.0"
+
+
+class OAuthClientProvider:
+    mcp_provider: MCPToolProvider
+
+    def __init__(self, provider_id: str, tenant_id: str, for_list: bool = False):
+        if for_list:
+            self.mcp_provider = MCPToolManageService.get_mcp_provider_by_provider_id(provider_id, tenant_id)
+        else:
+            self.mcp_provider = MCPToolManageService.get_mcp_provider_by_server_identifier(provider_id, tenant_id)
+
+    @property
+    def redirect_url(self) -> str:
+        """The URL to redirect the user agent to after authorization."""
+        return dify_config.CONSOLE_API_URL + "/console/api/mcp/oauth/callback"
+
+    @property
+    def client_metadata(self) -> OAuthClientMetadata:
+        """Metadata about this OAuth client."""
+        return OAuthClientMetadata(
+            redirect_uris=[self.redirect_url],
+            token_endpoint_auth_method="none",
+            grant_types=["authorization_code", "refresh_token"],
+            response_types=["code"],
+            client_name="Dify",
+            client_uri="https://github.com/langgenius/dify",
+        )
+
+    def client_information(self) -> Optional[OAuthClientInformation]:
+        """Loads information about this OAuth client."""
+        client_information = self.mcp_provider.decrypted_credentials.get("client_information", {})
+        if not client_information:
+            return None
+        return OAuthClientInformation.model_validate(client_information)
+
+    def save_client_information(self, client_information: OAuthClientInformationFull) -> None:
+        """Saves client information after dynamic registration."""
+        MCPToolManageService.update_mcp_provider_credentials(
+            self.mcp_provider,
+            {"client_information": client_information.model_dump()},
+        )
+
+    def tokens(self) -> Optional[OAuthTokens]:
+        """Loads any existing OAuth tokens for the current session."""
+        credentials = self.mcp_provider.decrypted_credentials
+        if not credentials:
+            return None
+        return OAuthTokens(
+            access_token=credentials.get("access_token", ""),
+            token_type=credentials.get("token_type", "Bearer"),
+            expires_in=int(credentials.get("expires_in", "3600") or 3600),
+            refresh_token=credentials.get("refresh_token", ""),
+        )
+
+    def save_tokens(self, tokens: OAuthTokens) -> None:
+        """Stores new OAuth tokens for the current session."""
+        # update mcp provider credentials
+        token_dict = tokens.model_dump()
+        MCPToolManageService.update_mcp_provider_credentials(self.mcp_provider, token_dict, authed=True)
+
+    def save_code_verifier(self, code_verifier: str) -> None:
+        """Saves a PKCE code verifier for the current session."""
+        MCPToolManageService.update_mcp_provider_credentials(self.mcp_provider, {"code_verifier": code_verifier})
+
+    def code_verifier(self) -> str:
+        """Loads the PKCE code verifier for the current session."""
+        # get code verifier from mcp provider credentials
+        return str(self.mcp_provider.decrypted_credentials.get("code_verifier", ""))
@@ -0,0 +1,361 @@
+import logging
+import queue
+from collections.abc import Generator
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from typing import Any, TypeAlias, final
+from urllib.parse import urljoin, urlparse
+
+import httpx
+from sseclient import SSEClient
+
+from core.mcp import types
+from core.mcp.error import MCPAuthError, MCPConnectionError
+from core.mcp.types import SessionMessage
+from core.mcp.utils import create_ssrf_proxy_mcp_http_client, ssrf_proxy_sse_connect
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_QUEUE_READ_TIMEOUT = 3
+
+
+@final
+class _StatusReady:
+    def __init__(self, endpoint_url: str):
+        self._endpoint_url = endpoint_url
+
+
+@final
+class _StatusError:
+    def __init__(self, exc: Exception):
+        self._exc = exc
+
+
+# Type aliases for better readability
+ReadQueue: TypeAlias = queue.Queue[SessionMessage | Exception | None]
+WriteQueue: TypeAlias = queue.Queue[SessionMessage | Exception | None]
+StatusQueue: TypeAlias = queue.Queue[_StatusReady | _StatusError]
+
+
+def remove_request_params(url: str) -> str:
+    """Remove request parameters from URL, keeping only the path."""
+    return urljoin(url, urlparse(url).path)
+
+
+class SSETransport:
+    """SSE client transport implementation."""
+
+    def __init__(
+        self,
+        url: str,
+        headers: dict[str, Any] | None = None,
+        timeout: float = 5.0,
+        sse_read_timeout: float = 5 * 60,
+    ) -> None:
+        """Initialize the SSE transport.
+
+        Args:
+            url: The SSE endpoint URL.
+            headers: Optional headers to include in requests.
+            timeout: HTTP timeout for regular operations.
+            sse_read_timeout: Timeout for SSE read operations.
+        """
+        self.url = url
+        self.headers = headers or {}
+        self.timeout = timeout
+        self.sse_read_timeout = sse_read_timeout
+        self.endpoint_url: str | None = None
+
+    def _validate_endpoint_url(self, endpoint_url: str) -> bool:
+        """Validate that the endpoint URL matches the connection origin.
+
+        Args:
+            endpoint_url: The endpoint URL to validate.
+
+        Returns:
+            True if valid, False otherwise.
+        """
+        url_parsed = urlparse(self.url)
+        endpoint_parsed = urlparse(endpoint_url)
+
+        return url_parsed.netloc == endpoint_parsed.netloc and url_parsed.scheme == endpoint_parsed.scheme
+
+    def _handle_endpoint_event(self, sse_data: str, status_queue: StatusQueue) -> None:
+        """Handle an 'endpoint' SSE event.
+
+        Args:
+            sse_data: The SSE event data.
+            status_queue: Queue to put status updates.
+        """
+        endpoint_url = urljoin(self.url, sse_data)
+        logger.info(f"Received endpoint URL: {endpoint_url}")
+
+        if not self._validate_endpoint_url(endpoint_url):
+            error_msg = f"Endpoint origin does not match connection origin: {endpoint_url}"
+            logger.error(error_msg)
+            status_queue.put(_StatusError(ValueError(error_msg)))
+            return
+
+        status_queue.put(_StatusReady(endpoint_url))
+
+    def _handle_message_event(self, sse_data: str, read_queue: ReadQueue) -> None:
+        """Handle a 'message' SSE event.
+
+        Args:
+            sse_data: The SSE event data.
+            read_queue: Queue to put parsed messages.
+        """
+        try:
+            message = types.JSONRPCMessage.model_validate_json(sse_data)
+            logger.debug(f"Received server message: {message}")
+            session_message = SessionMessage(message)
+            read_queue.put(session_message)
+        except Exception as exc:
+            logger.exception("Error parsing server message")
+            read_queue.put(exc)
+
+    def _handle_sse_event(self, sse, read_queue: ReadQueue, status_queue: StatusQueue) -> None:
+        """Handle a single SSE event.
+
+        Args:
+            sse: The SSE event object.
+            read_queue: Queue for message events.
+            status_queue: Queue for status events.
+        """
+        match sse.event:
+            case "endpoint":
+                self._handle_endpoint_event(sse.data, status_queue)
+            case "message":
+                self._handle_message_event(sse.data, read_queue)
+            case _:
+                logger.warning(f"Unknown SSE event: {sse.event}")
+
+    def sse_reader(self, event_source, read_queue: ReadQueue, status_queue: StatusQueue) -> None:
+        """Read and process SSE events.
+
+        Args:
+            event_source: The SSE event source.
+            read_queue: Queue to put received messages.
+            status_queue: Queue to put status updates.
+        """
+        try:
+            for sse in event_source.iter_sse():
+                self._handle_sse_event(sse, read_queue, status_queue)
+        except httpx.ReadError as exc:
+            logger.debug(f"SSE reader shutting down normally: {exc}")
+        except Exception as exc:
+            read_queue.put(exc)
+        finally:
+            read_queue.put(None)
+
+    def _send_message(self, client: httpx.Client, endpoint_url: str, message: SessionMessage) -> None:
+        """Send a single message to the server.
+
+        Args:
+            client: HTTP client to use.
+            endpoint_url: The endpoint URL to send to.
+            message: The message to send.
+        """
+        response = client.post(
+            endpoint_url,
+            json=message.message.model_dump(
+                by_alias=True,
+                mode="json",
+                exclude_none=True,
+            ),
+        )
+        response.raise_for_status()
+        logger.debug(f"Client message sent successfully: {response.status_code}")
+
+    def post_writer(self, client: httpx.Client, endpoint_url: str, write_queue: WriteQueue) -> None:
+        """Handle writing messages to the server.
+
+        Args:
+            client: HTTP client to use.
+            endpoint_url: The endpoint URL to send messages to.
+            write_queue: Queue to read messages from.
+        """
+        try:
+            while True:
+                try:
+                    message = write_queue.get(timeout=DEFAULT_QUEUE_READ_TIMEOUT)
+                    if message is None:
+                        break
+                    if isinstance(message, Exception):
+                        write_queue.put(message)
+                        continue
+
+                    self._send_message(client, endpoint_url, message)
+
+                except queue.Empty:
+                    continue
+        except httpx.ReadError as exc:
+            logger.debug(f"Post writer shutting down normally: {exc}")
+        except Exception as exc:
+            logger.exception("Error writing messages")
+            write_queue.put(exc)
+        finally:
+            write_queue.put(None)
+
+    def _wait_for_endpoint(self, status_queue: StatusQueue) -> str:
+        """Wait for the endpoint URL from the status queue.
+
+        Args:
+            status_queue: Queue to read status from.
+
+        Returns:
+            The endpoint URL.
+
+        Raises:
+            ValueError: If endpoint URL is not received or there's an error.
+        """
+        try:
+            status = status_queue.get(timeout=1)
+        except queue.Empty:
+            raise ValueError("failed to get endpoint URL")
+
+        if isinstance(status, _StatusReady):
+            return status._endpoint_url
+        elif isinstance(status, _StatusError):
+            raise status._exc
+        else:
+            raise ValueError("failed to get endpoint URL")
+
+    def connect(
+        self,
+        executor: ThreadPoolExecutor,
+        client: httpx.Client,
+        event_source,
+    ) -> tuple[ReadQueue, WriteQueue]:
+        """Establish connection and start worker threads.
+
+        Args:
+            executor: Thread pool executor.
+            client: HTTP client.
+            event_source: SSE event source.
+
+        Returns:
+            Tuple of (read_queue, write_queue).
+        """
+        read_queue: ReadQueue = queue.Queue()
+        write_queue: WriteQueue = queue.Queue()
+        status_queue: StatusQueue = queue.Queue()
+
+        # Start SSE reader thread
+        executor.submit(self.sse_reader, event_source, read_queue, status_queue)
+
+        # Wait for endpoint URL
+        endpoint_url = self._wait_for_endpoint(status_queue)
+        self.endpoint_url = endpoint_url
+
+        # Start post writer thread
+        executor.submit(self.post_writer, client, endpoint_url, write_queue)
+
+        return read_queue, write_queue
+
+
+@contextmanager
+def sse_client(
+    url: str,
+    headers: dict[str, Any] | None = None,
+    timeout: float = 5.0,
+    sse_read_timeout: float = 5 * 60,
+) -> Generator[tuple[ReadQueue, WriteQueue], None, None]:
+    """
+    Client transport for SSE.
+    `sse_read_timeout` determines how long (in seconds) the client will wait for a new
+    event before disconnecting. All other HTTP operations are controlled by `timeout`.
+
+    Args:
+        url: The SSE endpoint URL.
+        headers: Optional headers to include in requests.
+        timeout: HTTP timeout for regular operations.
+        sse_read_timeout: Timeout for SSE read operations.
+
+    Yields:
+        Tuple of (read_queue, write_queue) for message communication.
+    """
+    transport = SSETransport(url, headers, timeout, sse_read_timeout)
+
+    read_queue: ReadQueue | None = None
+    write_queue: WriteQueue | None = None
+
+    with ThreadPoolExecutor() as executor:
+        try:
+            with create_ssrf_proxy_mcp_http_client(headers=transport.headers) as client:
+                with ssrf_proxy_sse_connect(
+                    url, timeout=httpx.Timeout(timeout, read=sse_read_timeout), client=client
+                ) as event_source:
+                    event_source.response.raise_for_status()
+
+                    read_queue, write_queue = transport.connect(executor, client, event_source)
+
+                    yield read_queue, write_queue
+
+        except httpx.HTTPStatusError as exc:
+            if exc.response.status_code == 401:
+                raise MCPAuthError()
+            raise MCPConnectionError()
+        except Exception:
+            logger.exception("Error connecting to SSE endpoint")
+            raise
+        finally:
+            # Clean up queues
+            if read_queue:
+                read_queue.put(None)
+            if write_queue:
+                write_queue.put(None)
+
+
+def send_message(http_client: httpx.Client, endpoint_url: str, session_message: SessionMessage) -> None:
+    """
+    Send a message to the server using the provided HTTP client.
+
+    Args:
+        http_client: The HTTP client to use for sending
+        endpoint_url: The endpoint URL to send the message to
+        session_message: The message to send
+    """
+    try:
+        response = http_client.post(
+            endpoint_url,
+            json=session_message.message.model_dump(
+                by_alias=True,
+                mode="json",
+                exclude_none=True,
+            ),
+        )
+        response.raise_for_status()
+        logger.debug(f"Client message sent successfully: {response.status_code}")
+    except Exception as exc:
+        logger.exception("Error sending message")
+        raise
+
+
+def read_messages(
+    sse_client: SSEClient,
+) -> Generator[SessionMessage | Exception, None, None]:
+    """
+    Read messages from the SSE client.
+
+    Args:
+        sse_client: The SSE client to read from
+
+    Yields:
+        SessionMessage or Exception for each event received
+    """
+    try:
+        for sse in sse_client.events():
+            if sse.event == "message":
+                try:
+                    message = types.JSONRPCMessage.model_validate_json(sse.data)
+                    logger.debug(f"Received server message: {message}")
+                    yield SessionMessage(message)
+                except Exception as exc:
+                    logger.exception("Error parsing server message")
+                    yield exc
+            else:
+                logger.warning(f"Unknown SSE event: {sse.event}")
+    except Exception as exc:
+        logger.exception("Error reading SSE messages")
+        yield exc
@@ -0,0 +1,476 @@
+"""
+StreamableHTTP Client Transport Module
+
+This module implements the StreamableHTTP transport for MCP clients,
+providing support for HTTP POST requests with optional SSE streaming responses
+and session management.
+"""
+
+import logging
+import queue
+from collections.abc import Callable, Generator
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Any, cast
+
+import httpx
+from httpx_sse import EventSource, ServerSentEvent
+
+from core.mcp.types import (
+    ClientMessageMetadata,
+    ErrorData,
+    JSONRPCError,
+    JSONRPCMessage,
+    JSONRPCNotification,
+    JSONRPCRequest,
+    JSONRPCResponse,
+    RequestId,
+    SessionMessage,
+)
+from core.mcp.utils import create_ssrf_proxy_mcp_http_client, ssrf_proxy_sse_connect
+
+logger = logging.getLogger(__name__)
+
+
+SessionMessageOrError = SessionMessage | Exception | None
+# Queue types with clearer names for their roles
+ServerToClientQueue = queue.Queue[SessionMessageOrError]  # Server to client messages
+ClientToServerQueue = queue.Queue[SessionMessage | None]  # Client to server messages
+GetSessionIdCallback = Callable[[], str | None]
+
+MCP_SESSION_ID = "mcp-session-id"
+LAST_EVENT_ID = "last-event-id"
+CONTENT_TYPE = "content-type"
+ACCEPT = "Accept"
+
+
+JSON = "application/json"
+SSE = "text/event-stream"
+
+DEFAULT_QUEUE_READ_TIMEOUT = 3
+
+
+class StreamableHTTPError(Exception):
+    """Base exception for StreamableHTTP transport errors."""
+
+    pass
+
+
+class ResumptionError(StreamableHTTPError):
+    """Raised when resumption request is invalid."""
+
+    pass
+
+
+@dataclass
+class RequestContext:
+    """Context for a request operation."""
+
+    client: httpx.Client
+    headers: dict[str, str]
+    session_id: str | None
+    session_message: SessionMessage
+    metadata: ClientMessageMetadata | None
+    server_to_client_queue: ServerToClientQueue  # Renamed for clarity
+    sse_read_timeout: timedelta
+
+
+class StreamableHTTPTransport:
+    """StreamableHTTP client transport implementation."""
+
+    def __init__(
+        self,
+        url: str,
+        headers: dict[str, Any] | None = None,
+        timeout: timedelta = timedelta(seconds=30),
+        sse_read_timeout: timedelta = timedelta(seconds=60 * 5),
+    ) -> None:
+        """Initialize the StreamableHTTP transport.
+
+        Args:
+            url: The endpoint URL.
+            headers: Optional headers to include in requests.
+            timeout: HTTP timeout for regular operations.
+            sse_read_timeout: Timeout for SSE read operations.
+        """
+        self.url = url
+        self.headers = headers or {}
+        self.timeout = timeout
+        self.sse_read_timeout = sse_read_timeout
+        self.session_id: str | None = None
+        self.request_headers = {
+            ACCEPT: f"{JSON}, {SSE}",
+            CONTENT_TYPE: JSON,
+            **self.headers,
+        }
+
+    def _update_headers_with_session(self, base_headers: dict[str, str]) -> dict[str, str]:
+        """Update headers with session ID if available."""
+        headers = base_headers.copy()
+        if self.session_id:
+            headers[MCP_SESSION_ID] = self.session_id
+        return headers
+
+    def _is_initialization_request(self, message: JSONRPCMessage) -> bool:
+        """Check if the message is an initialization request."""
+        return isinstance(message.root, JSONRPCRequest) and message.root.method == "initialize"
+
+    def _is_initialized_notification(self, message: JSONRPCMessage) -> bool:
+        """Check if the message is an initialized notification."""
+        return isinstance(message.root, JSONRPCNotification) and message.root.method == "notifications/initialized"
+
+    def _maybe_extract_session_id_from_response(
+        self,
+        response: httpx.Response,
+    ) -> None:
+        """Extract and store session ID from response headers."""
+        new_session_id = response.headers.get(MCP_SESSION_ID)
+        if new_session_id:
+            self.session_id = new_session_id
+            logger.info(f"Received session ID: {self.session_id}")
+
+    def _handle_sse_event(
+        self,
+        sse: ServerSentEvent,
+        server_to_client_queue: ServerToClientQueue,
+        original_request_id: RequestId | None = None,
+        resumption_callback: Callable[[str], None] | None = None,
+    ) -> bool:
+        """Handle an SSE event, returning True if the response is complete."""
+        if sse.event == "message":
+            try:
+                message = JSONRPCMessage.model_validate_json(sse.data)
+                logger.debug(f"SSE message: {message}")
+
+                # If this is a response and we have original_request_id, replace it
+                if original_request_id is not None and isinstance(message.root, JSONRPCResponse | JSONRPCError):
+                    message.root.id = original_request_id
+
+                session_message = SessionMessage(message)
+                # Put message in queue that goes to client
+                server_to_client_queue.put(session_message)
+
+                # Call resumption token callback if we have an ID
+                if sse.id and resumption_callback:
+                    resumption_callback(sse.id)
+
+                # If this is a response or error return True indicating completion
+                # Otherwise, return False to continue listening
+                return isinstance(message.root, JSONRPCResponse | JSONRPCError)
+
+            except Exception as exc:
+                # Put exception in queue that goes to client
+                server_to_client_queue.put(exc)
+                return False
+        elif sse.event == "ping":
+            logger.debug("Received ping event")
+            return False
+        else:
+            logger.warning(f"Unknown SSE event: {sse.event}")
+            return False
+
+    def handle_get_stream(
+        self,
+        client: httpx.Client,
+        server_to_client_queue: ServerToClientQueue,
+    ) -> None:
+        """Handle GET stream for server-initiated messages."""
+        try:
+            if not self.session_id:
+                return
+
+            headers = self._update_headers_with_session(self.request_headers)
+
+            with ssrf_proxy_sse_connect(
+                self.url,
+                headers=headers,
+                timeout=httpx.Timeout(self.timeout.seconds, read=self.sse_read_timeout.seconds),
+                client=client,
+                method="GET",
+            ) as event_source:
+                event_source.response.raise_for_status()
+                logger.debug("GET SSE connection established")
+
+                for sse in event_source.iter_sse():
+                    self._handle_sse_event(sse, server_to_client_queue)
+
+        except Exception as exc:
+            logger.debug(f"GET stream error (non-fatal): {exc}")
+
+    def _handle_resumption_request(self, ctx: RequestContext) -> None:
+        """Handle a resumption request using GET with SSE."""
+        headers = self._update_headers_with_session(ctx.headers)
+        if ctx.metadata and ctx.metadata.resumption_token:
+            headers[LAST_EVENT_ID] = ctx.metadata.resumption_token
+        else:
+            raise ResumptionError("Resumption request requires a resumption token")
+
+        # Extract original request ID to map responses
+        original_request_id = None
+        if isinstance(ctx.session_message.message.root, JSONRPCRequest):
+            original_request_id = ctx.session_message.message.root.id
+
+        with ssrf_proxy_sse_connect(
+            self.url,
+            headers=headers,
+            timeout=httpx.Timeout(self.timeout.seconds, read=ctx.sse_read_timeout.seconds),
+            client=ctx.client,
+            method="GET",
+        ) as event_source:
+            event_source.response.raise_for_status()
+            logger.debug("Resumption GET SSE connection established")
+
+            for sse in event_source.iter_sse():
+                is_complete = self._handle_sse_event(
+                    sse,
+                    ctx.server_to_client_queue,
+                    original_request_id,
+                    ctx.metadata.on_resumption_token_update if ctx.metadata else None,
+                )
+                if is_complete:
+                    break
+
+    def _handle_post_request(self, ctx: RequestContext) -> None:
+        """Handle a POST request with response processing."""
+        headers = self._update_headers_with_session(ctx.headers)
+        message = ctx.session_message.message
+        is_initialization = self._is_initialization_request(message)
+
+        with ctx.client.stream(
+            "POST",
+            self.url,
+            json=message.model_dump(by_alias=True, mode="json", exclude_none=True),
+            headers=headers,
+        ) as response:
+            if response.status_code == 202:
+                logger.debug("Received 202 Accepted")
+                return
+
+            if response.status_code == 404:
+                if isinstance(message.root, JSONRPCRequest):
+                    self._send_session_terminated_error(
+                        ctx.server_to_client_queue,
+                        message.root.id,
+                    )
+                return
+
+            response.raise_for_status()
+            if is_initialization:
+                self._maybe_extract_session_id_from_response(response)
+
+            content_type = cast(str, response.headers.get(CONTENT_TYPE, "").lower())
+
+            if content_type.startswith(JSON):
+                self._handle_json_response(response, ctx.server_to_client_queue)
+            elif content_type.startswith(SSE):
+                self._handle_sse_response(response, ctx)
+            else:
+                self._handle_unexpected_content_type(
+                    content_type,
+                    ctx.server_to_client_queue,
+                )
+
+    def _handle_json_response(
+        self,
+        response: httpx.Response,
+        server_to_client_queue: ServerToClientQueue,
+    ) -> None:
+        """Handle JSON response from the server."""
+        try:
+            content = response.read()
+            message = JSONRPCMessage.model_validate_json(content)
+            session_message = SessionMessage(message)
+            server_to_client_queue.put(session_message)
+        except Exception as exc:
+            server_to_client_queue.put(exc)
+
+    def _handle_sse_response(self, response: httpx.Response, ctx: RequestContext) -> None:
+        """Handle SSE response from the server."""
+        try:
+            event_source = EventSource(response)
+            for sse in event_source.iter_sse():
+                is_complete = self._handle_sse_event(
+                    sse,
+                    ctx.server_to_client_queue,
+                    resumption_callback=(ctx.metadata.on_resumption_token_update if ctx.metadata else None),
+                )
+                if is_complete:
+                    break
+        except Exception as e:
+            ctx.server_to_client_queue.put(e)
+
+    def _handle_unexpected_content_type(
+        self,
+        content_type: str,
+        server_to_client_queue: ServerToClientQueue,
+    ) -> None:
+        """Handle unexpected content type in response."""
+        error_msg = f"Unexpected content type: {content_type}"
+        logger.error(error_msg)
+        server_to_client_queue.put(ValueError(error_msg))
+
+    def _send_session_terminated_error(
+        self,
+        server_to_client_queue: ServerToClientQueue,
+        request_id: RequestId,
+    ) -> None:
+        """Send a session terminated error response."""
+        jsonrpc_error = JSONRPCError(
+            jsonrpc="2.0",
+            id=request_id,
+            error=ErrorData(code=32600, message="Session terminated by server"),
+        )
+        session_message = SessionMessage(JSONRPCMessage(jsonrpc_error))
+        server_to_client_queue.put(session_message)
+
+    def post_writer(
+        self,
+        client: httpx.Client,
+        client_to_server_queue: ClientToServerQueue,
+        server_to_client_queue: ServerToClientQueue,
+        start_get_stream: Callable[[], None],
+    ) -> None:
+        """Handle writing requests to the server.
+
+        This method processes messages from the client_to_server_queue and sends them to the server.
+        Responses are written to the server_to_client_queue.
+        """
+        while True:
+            try:
+                # Read message from client queue with timeout to check stop_event periodically
+                session_message = client_to_server_queue.get(timeout=DEFAULT_QUEUE_READ_TIMEOUT)
+                if session_message is None:
+                    break
+
+                message = session_message.message
+                metadata = (
+                    session_message.metadata if isinstance(session_message.metadata, ClientMessageMetadata) else None
+                )
+
+                # Check if this is a resumption request
+                is_resumption = bool(metadata and metadata.resumption_token)
+
+                logger.debug(f"Sending client message: {message}")
+
+                # Handle initialized notification
+                if self._is_initialized_notification(message):
+                    start_get_stream()
+
+                ctx = RequestContext(
+                    client=client,
+                    headers=self.request_headers,
+                    session_id=self.session_id,
+                    session_message=session_message,
+                    metadata=metadata,
+                    server_to_client_queue=server_to_client_queue,  # Queue to write responses to client
+                    sse_read_timeout=self.sse_read_timeout,
+                )
+
+                if is_resumption:
+                    self._handle_resumption_request(ctx)
+                else:
+                    self._handle_post_request(ctx)
+            except queue.Empty:
+                continue
+            except Exception as exc:
+                server_to_client_queue.put(exc)
+
+    def terminate_session(self, client: httpx.Client) -> None:
+        """Terminate the session by sending a DELETE request."""
+        if not self.session_id:
+            return
+
+        try:
+            headers = self._update_headers_with_session(self.request_headers)
+            response = client.delete(self.url, headers=headers)
+
+            if response.status_code == 405:
+                logger.debug("Server does not allow session termination")
+            elif response.status_code != 200:
+                logger.warning(f"Session termination failed: {response.status_code}")
+        except Exception as exc:
+            logger.warning(f"Session termination failed: {exc}")
+
+    def get_session_id(self) -> str | None:
+        """Get the current session ID."""
+        return self.session_id
+
+
+@contextmanager
+def streamablehttp_client(
+    url: str,
+    headers: dict[str, Any] | None = None,
+    timeout: timedelta = timedelta(seconds=30),
+    sse_read_timeout: timedelta = timedelta(seconds=60 * 5),
+    terminate_on_close: bool = True,
+) -> Generator[
+    tuple[
+        ServerToClientQueue,  # Queue for receiving messages FROM server
+        ClientToServerQueue,  # Queue for sending messages TO server
+        GetSessionIdCallback,
+    ],
+    None,
+    None,
+]:
+    """
+    Client transport for StreamableHTTP.
+
+    `sse_read_timeout` determines how long (in seconds) the client will wait for a new
+    event before disconnecting. All other HTTP operations are controlled by `timeout`.
+
+    Yields:
+        Tuple containing:
+            - server_to_client_queue: Queue for reading messages FROM the server
+            - client_to_server_queue: Queue for sending messages TO the server
+            - get_session_id_callback: Function to retrieve the current session ID
+    """
+    transport = StreamableHTTPTransport(url, headers, timeout, sse_read_timeout)
+
+    # Create queues with clear directional meaning
+    server_to_client_queue: ServerToClientQueue = queue.Queue()  # For messages FROM server TO client
+    client_to_server_queue: ClientToServerQueue = queue.Queue()  # For messages FROM client TO server
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        try:
+            with create_ssrf_proxy_mcp_http_client(
+                headers=transport.request_headers,
+                timeout=httpx.Timeout(transport.timeout.seconds, read=transport.sse_read_timeout.seconds),
+            ) as client:
+                # Define callbacks that need access to thread pool
+                def start_get_stream() -> None:
+                    """Start a worker thread to handle server-initiated messages."""
+                    executor.submit(transport.handle_get_stream, client, server_to_client_queue)
+
+                # Start the post_writer worker thread
+                executor.submit(
+                    transport.post_writer,
+                    client,
+                    client_to_server_queue,  # Queue for messages FROM client TO server
+                    server_to_client_queue,  # Queue for messages FROM server TO client
+                    start_get_stream,
+                )
+
+                try:
+                    yield (
+                        server_to_client_queue,  # Queue for receiving messages FROM server
+                        client_to_server_queue,  # Queue for sending messages TO server
+                        transport.get_session_id,
+                    )
+                finally:
+                    if transport.session_id and terminate_on_close:
+                        transport.terminate_session(client)
+
+                    # Signal threads to stop
+                    client_to_server_queue.put(None)
+        finally:
+            # Clear any remaining items and add None sentinel to unblock any waiting threads
+            try:
+                while not client_to_server_queue.empty():
+                    client_to_server_queue.get_nowait()
+            except queue.Empty:
+                pass
+
+            client_to_server_queue.put(None)
+            server_to_client_queue.put(None)
@@ -0,0 +1,19 @@
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar
+
+from core.mcp.session.base_session import BaseSession
+from core.mcp.types import LATEST_PROTOCOL_VERSION, RequestId, RequestParams
+
+SUPPORTED_PROTOCOL_VERSIONS: list[str] = ["2024-11-05", LATEST_PROTOCOL_VERSION]
+
+
+SessionT = TypeVar("SessionT", bound=BaseSession[Any, Any, Any, Any, Any])
+LifespanContextT = TypeVar("LifespanContextT")
+
+
+@dataclass
+class RequestContext(Generic[SessionT, LifespanContextT]):
+    request_id: RequestId
+    meta: RequestParams.Meta | None
+    session: SessionT
+    lifespan_context: LifespanContextT
@@ -0,0 +1,10 @@
+class MCPError(Exception):
+    pass
+
+
+class MCPConnectionError(MCPError):
+    pass
+
+
+class MCPAuthError(MCPConnectionError):
+    pass
@@ -0,0 +1,150 @@
+import logging
+from collections.abc import Callable
+from contextlib import AbstractContextManager, ExitStack
+from types import TracebackType
+from typing import Any, Optional, cast
+from urllib.parse import urlparse
+
+from core.mcp.client.sse_client import sse_client
+from core.mcp.client.streamable_client import streamablehttp_client
+from core.mcp.error import MCPAuthError, MCPConnectionError
+from core.mcp.session.client_session import ClientSession
+from core.mcp.types import Tool
+
+logger = logging.getLogger(__name__)
+
+
+class MCPClient:
+    def __init__(
+        self,
+        server_url: str,
+        provider_id: str,
+        tenant_id: str,
+        authed: bool = True,
+        authorization_code: Optional[str] = None,
+        for_list: bool = False,
+    ):
+        # Initialize info
+        self.provider_id = provider_id
+        self.tenant_id = tenant_id
+        self.client_type = "streamable"
+        self.server_url = server_url
+
+        # Authentication info
+        self.authed = authed
+        self.authorization_code = authorization_code
+        if authed:
+            from core.mcp.auth.auth_provider import OAuthClientProvider
+
+            self.provider = OAuthClientProvider(self.provider_id, self.tenant_id, for_list=for_list)
+            self.token = self.provider.tokens()
+
+        # Initialize session and client objects
+        self._session: Optional[ClientSession] = None
+        self._streams_context: Optional[AbstractContextManager[Any]] = None
+        self._session_context: Optional[ClientSession] = None
+        self.exit_stack = ExitStack()
+
+        # Whether the client has been initialized
+        self._initialized = False
+
+    def __enter__(self):
+        self._initialize()
+        self._initialized = True
+        return self
+
+    def __exit__(
+        self, exc_type: Optional[type], exc_value: Optional[BaseException], traceback: Optional[TracebackType]
+    ):
+        self.cleanup()
+
+    def _initialize(
+        self,
+    ):
+        """Initialize the client with fallback to SSE if streamable connection fails"""
+        connection_methods: dict[str, Callable[..., AbstractContextManager[Any]]] = {
+            "mcp": streamablehttp_client,
+            "sse": sse_client,
+        }
+
+        parsed_url = urlparse(self.server_url)
+        path = parsed_url.path
+        method_name = path.rstrip("/").split("/")[-1] if path else ""
+        try:
+            client_factory = connection_methods[method_name]
+            self.connect_server(client_factory, method_name)
+        except KeyError:
+            try:
+                self.connect_server(sse_client, "sse")
+            except MCPConnectionError:
+                self.connect_server(streamablehttp_client, "mcp")
+
+    def connect_server(
+        self, client_factory: Callable[..., AbstractContextManager[Any]], method_name: str, first_try: bool = True
+    ):
+        from core.mcp.auth.auth_flow import auth
+
+        try:
+            headers = (
+                {"Authorization": f"{self.token.token_type.capitalize()} {self.token.access_token}"}
+                if self.authed and self.token
+                else {}
+            )
+            self._streams_context = client_factory(url=self.server_url, headers=headers)
+            if self._streams_context is None:
+                raise MCPConnectionError("Failed to create connection context")
+
+            # Use exit_stack to manage context managers properly
+            if method_name == "mcp":
+                read_stream, write_stream, _ = self.exit_stack.enter_context(self._streams_context)
+                streams = (read_stream, write_stream)
+            else:  # sse_client
+                streams = self.exit_stack.enter_context(self._streams_context)
+
+            self._session_context = ClientSession(*streams)
+            self._session = self.exit_stack.enter_context(self._session_context)
+            session = cast(ClientSession, self._session)
+            session.initialize()
+            return
+
+        except MCPAuthError:
+            if not self.authed:
+                raise
+            try:
+                auth(self.provider, self.server_url, self.authorization_code)
+            except Exception as e:
+                raise ValueError(f"Failed to authenticate: {e}")
+            self.token = self.provider.tokens()
+            if first_try:
+                return self.connect_server(client_factory, method_name, first_try=False)
+
+        except MCPConnectionError:
+            raise
+
+    def list_tools(self) -> list[Tool]:
+        """Connect to an MCP server running with SSE transport"""
+        # List available tools to verify connection
+        if not self._initialized or not self._session:
+            raise ValueError("Session not initialized.")
+        response = self._session.list_tools()
+        tools = response.tools
+        return tools
+
+    def invoke_tool(self, tool_name: str, tool_args: dict):
+        """Call a tool"""
+        if not self._initialized or not self._session:
+            raise ValueError("Session not initialized.")
+        return self._session.call_tool(tool_name, tool_args)
+
+    def cleanup(self):
+        """Clean up resources"""
+        try:
+            # ExitStack will handle proper cleanup of all managed context managers
+            self.exit_stack.close()
+            self._session = None
+            self._session_context = None
+            self._streams_context = None
+            self._initialized = False
+        except Exception as e:
+            logging.exception("Error during cleanup")
+            raise ValueError(f"Error during cleanup: {e}")
@@ -0,0 +1,224 @@
+import json
+import logging
+from collections.abc import Mapping
+from typing import Any, cast
+
+from configs import dify_config
+from controllers.web.passport import generate_session_id
+from core.app.app_config.entities import VariableEntity, VariableEntityType
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
+from core.mcp import types
+from core.mcp.types import INTERNAL_ERROR, INVALID_PARAMS, METHOD_NOT_FOUND
+from core.mcp.utils import create_mcp_error_response
+from core.model_runtime.utils.encoders import jsonable_encoder
+from extensions.ext_database import db
+from models.model import App, AppMCPServer, AppMode, EndUser
+from services.app_generate_service import AppGenerateService
+
+"""
+Apply to MCP HTTP streamable server with stateless http
+"""
+logger = logging.getLogger(__name__)
+
+
+class MCPServerStreamableHTTPRequestHandler:
+    def __init__(
+        self, app: App, request: types.ClientRequest | types.ClientNotification, user_input_form: list[VariableEntity]
+    ):
+        self.app = app
+        self.request = request
+        mcp_server = db.session.query(AppMCPServer).filter(AppMCPServer.app_id == self.app.id).first()
+        if not mcp_server:
+            raise ValueError("MCP server not found")
+        self.mcp_server: AppMCPServer = mcp_server
+        self.end_user = self.retrieve_end_user()
+        self.user_input_form = user_input_form
+
+    @property
+    def request_type(self):
+        return type(self.request.root)
+
+    @property
+    def parameter_schema(self):
+        parameters, required = self._convert_input_form_to_parameters(self.user_input_form)
+        if self.app.mode in {AppMode.COMPLETION.value, AppMode.WORKFLOW.value}:
+            return {
+                "type": "object",
+                "properties": parameters,
+                "required": required,
+            }
+        return {
+            "type": "object",
+            "properties": {
+                "query": {"type": "string", "description": "User Input/Question content"},
+                **parameters,
+            },
+            "required": ["query", *required],
+        }
+
+    @property
+    def capabilities(self):
+        return types.ServerCapabilities(
+            tools=types.ToolsCapability(listChanged=False),
+        )
+
+    def response(self, response: types.Result | str):
+        if isinstance(response, str):
+            sse_content = f"event: ping\ndata: {response}\n\n".encode()
+            yield sse_content
+            return
+        json_response = types.JSONRPCResponse(
+            jsonrpc="2.0",
+            id=(self.request.root.model_extra or {}).get("id", 1),
+            result=response.model_dump(by_alias=True, mode="json", exclude_none=True),
+        )
+        json_data = json.dumps(jsonable_encoder(json_response))
+
+        sse_content = f"event: message\ndata: {json_data}\n\n".encode()
+
+        yield sse_content
+
+    def error_response(self, code: int, message: str, data=None):
+        request_id = (self.request.root.model_extra or {}).get("id", 1) or 1
+        return create_mcp_error_response(request_id, code, message, data)
+
+    def handle(self):
+        handle_map = {
+            types.InitializeRequest: self.initialize,
+            types.ListToolsRequest: self.list_tools,
+            types.CallToolRequest: self.invoke_tool,
+            types.InitializedNotification: self.handle_notification,
+        }
+        try:
+            if self.request_type in handle_map:
+                return self.response(handle_map[self.request_type]())
+            else:
+                return self.error_response(METHOD_NOT_FOUND, f"Method not found: {self.request_type}")
+        except ValueError as e:
+            logger.exception("Invalid params")
+            return self.error_response(INVALID_PARAMS, str(e))
+        except Exception as e:
+            logger.exception("Internal server error")
+            return self.error_response(INTERNAL_ERROR, f"Internal server error: {str(e)}")
+
+    def handle_notification(self):
+        return "ping"
+
+    def initialize(self):
+        request = cast(types.InitializeRequest, self.request.root)
+        client_info = request.params.clientInfo
+        clinet_name = f"{client_info.name}@{client_info.version}"
+        if not self.end_user:
+            end_user = EndUser(
+                tenant_id=self.app.tenant_id,
+                app_id=self.app.id,
+                type="mcp",
+                name=clinet_name,
+                session_id=generate_session_id(),
+                external_user_id=self.mcp_server.id,
+            )
+            db.session.add(end_user)
+            db.session.commit()
+        return types.InitializeResult(
+            protocolVersion=types.SERVER_LATEST_PROTOCOL_VERSION,
+            capabilities=self.capabilities,
+            serverInfo=types.Implementation(name="Dify", version=dify_config.project.version),
+            instructions=self.mcp_server.description,
+        )
+
+    def list_tools(self):
+        if not self.end_user:
+            raise ValueError("User not found")
+        return types.ListToolsResult(
+            tools=[
+                types.Tool(
+                    name=self.app.name,
+                    description=self.mcp_server.description,
+                    inputSchema=self.parameter_schema,
+                )
+            ],
+        )
+
+    def invoke_tool(self):
+        if not self.end_user:
+            raise ValueError("User not found")
+        request = cast(types.CallToolRequest, self.request.root)
+        args = request.params.arguments
+        if not args:
+            raise ValueError("No arguments provided")
+        if self.app.mode in {AppMode.WORKFLOW.value}:
+            args = {"inputs": args}
+        elif self.app.mode in {AppMode.COMPLETION.value}:
+            args = {"query": "", "inputs": args}
+        else:
+            args = {"query": args["query"], "inputs": {k: v for k, v in args.items() if k != "query"}}
+        response = AppGenerateService.generate(
+            self.app,
+            self.end_user,
+            args,
+            InvokeFrom.SERVICE_API,
+            streaming=self.app.mode == AppMode.AGENT_CHAT.value,
+        )
+        answer = ""
+        if isinstance(response, RateLimitGenerator):
+            for item in response.generator:
+                data = item
+                if isinstance(data, str) and data.startswith("data: "):
+                    try:
+                        json_str = data[6:].strip()
+                        parsed_data = json.loads(json_str)
+                        if parsed_data.get("event") == "agent_thought":
+                            answer += parsed_data.get("thought", "")
+                    except json.JSONDecodeError:
+                        continue
+        if isinstance(response, Mapping):
+            if self.app.mode in {
+                AppMode.ADVANCED_CHAT.value,
+                AppMode.COMPLETION.value,
+                AppMode.CHAT.value,
+                AppMode.AGENT_CHAT.value,
+            }:
+                answer = response["answer"]
+            elif self.app.mode in {AppMode.WORKFLOW.value}:
+                answer = json.dumps(response["data"]["outputs"], ensure_ascii=False)
+            else:
+                raise ValueError("Invalid app mode")
+            # Not support image yet
+        return types.CallToolResult(content=[types.TextContent(text=answer, type="text")])
+
+    def retrieve_end_user(self):
+        return (
+            db.session.query(EndUser)
+            .filter(EndUser.external_user_id == self.mcp_server.id, EndUser.type == "mcp")
+            .first()
+        )
+
+    def _convert_input_form_to_parameters(self, user_input_form: list[VariableEntity]):
+        parameters: dict[str, dict[str, Any]] = {}
+        required = []
+        for item in user_input_form:
+            parameters[item.variable] = {}
+            if item.type in (
+                VariableEntityType.FILE,
+                VariableEntityType.FILE_LIST,
+                VariableEntityType.EXTERNAL_DATA_TOOL,
+            ):
+                continue
+            if item.required:
+                required.append(item.variable)
+            # if the workflow republished, the parameters not changed
+            # we should not raise error here
+            try:
+                description = self.mcp_server.parameters_dict[item.variable]
+            except KeyError:
+                description = ""
+            parameters[item.variable]["description"] = description
+            if item.type in (VariableEntityType.TEXT_INPUT, VariableEntityType.PARAGRAPH):
+                parameters[item.variable]["type"] = "string"
+            elif item.type == VariableEntityType.SELECT:
+                parameters[item.variable]["type"] = "string"
+                parameters[item.variable]["enum"] = item.options
+            elif item.type == VariableEntityType.NUMBER:
+                parameters[item.variable]["type"] = "float"
+        return parameters, required
@@ -0,0 +1,397 @@
+import logging
+import queue
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import ExitStack
+from datetime import timedelta
+from types import TracebackType
+from typing import Any, Generic, Self, TypeVar
+
+from httpx import HTTPStatusError
+from pydantic import BaseModel
+
+from core.mcp.error import MCPAuthError, MCPConnectionError
+from core.mcp.types import (
+    CancelledNotification,
+    ClientNotification,
+    ClientRequest,
+    ClientResult,
+    ErrorData,
+    JSONRPCError,
+    JSONRPCMessage,
+    JSONRPCNotification,
+    JSONRPCRequest,
+    JSONRPCResponse,
+    MessageMetadata,
+    RequestId,
+    RequestParams,
+    ServerMessageMetadata,
+    ServerNotification,
+    ServerRequest,
+    ServerResult,
+    SessionMessage,
+)
+
+SendRequestT = TypeVar("SendRequestT", ClientRequest, ServerRequest)
+SendResultT = TypeVar("SendResultT", ClientResult, ServerResult)
+SendNotificationT = TypeVar("SendNotificationT", ClientNotification, ServerNotification)
+ReceiveRequestT = TypeVar("ReceiveRequestT", ClientRequest, ServerRequest)
+ReceiveResultT = TypeVar("ReceiveResultT", bound=BaseModel)
+ReceiveNotificationT = TypeVar("ReceiveNotificationT", ClientNotification, ServerNotification)
+DEFAULT_RESPONSE_READ_TIMEOUT = 1.0
+
+
+class RequestResponder(Generic[ReceiveRequestT, SendResultT]):
+    """Handles responding to MCP requests and manages request lifecycle.
+
+    This class MUST be used as a context manager to ensure proper cleanup and
+    cancellation handling:
+
+    Example:
+        with request_responder as resp:
+            resp.respond(result)
+
+    The context manager ensures:
+    1. Proper cancellation scope setup and cleanup
+    2. Request completion tracking
+    3. Cleanup of in-flight requests
+    """
+
+    request: ReceiveRequestT
+    _session: Any
+    _on_complete: Callable[["RequestResponder[ReceiveRequestT, SendResultT]"], Any]
+
+    def __init__(
+        self,
+        request_id: RequestId,
+        request_meta: RequestParams.Meta | None,
+        request: ReceiveRequestT,
+        session: """BaseSession[
+            SendRequestT,
+            SendNotificationT,
+            SendResultT,
+            ReceiveRequestT,
+            ReceiveNotificationT
+        ]""",
+        on_complete: Callable[["RequestResponder[ReceiveRequestT, SendResultT]"], Any],
+    ) -> None:
+        self.request_id = request_id
+        self.request_meta = request_meta
+        self.request = request
+        self._session = session
+        self._completed = False
+        self._on_complete = on_complete
+        self._entered = False  # Track if we're in a context manager
+
+    def __enter__(self) -> "RequestResponder[ReceiveRequestT, SendResultT]":
+        """Enter the context manager, enabling request cancellation tracking."""
+        self._entered = True
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: TracebackType | None,
+    ) -> None:
+        """Exit the context manager, performing cleanup and notifying completion."""
+        try:
+            if self._completed:
+                self._on_complete(self)
+        finally:
+            self._entered = False
+
+    def respond(self, response: SendResultT | ErrorData) -> None:
+        """Send a response for this request.
+
+        Must be called within a context manager block.
+        Raises:
+            RuntimeError: If not used within a context manager
+            AssertionError: If request was already responded to
+        """
+        if not self._entered:
+            raise RuntimeError("RequestResponder must be used as a context manager")
+        assert not self._completed, "Request already responded to"
+
+        self._completed = True
+
+        self._session._send_response(request_id=self.request_id, response=response)
+
+    def cancel(self) -> None:
+        """Cancel this request and mark it as completed."""
+        if not self._entered:
+            raise RuntimeError("RequestResponder must be used as a context manager")
+
+        self._completed = True  # Mark as completed so it's removed from in_flight
+        # Send an error response to indicate cancellation
+        self._session._send_response(
+            request_id=self.request_id,
+            response=ErrorData(code=0, message="Request cancelled", data=None),
+        )
+
+
+class BaseSession(
+    Generic[
+        SendRequestT,
+        SendNotificationT,
+        SendResultT,
+        ReceiveRequestT,
+        ReceiveNotificationT,
+    ],
+):
+    """
+    Implements an MCP "session" on top of read/write streams, including features
+    like request/response linking, notifications, and progress.
+
+    This class is a context manager that automatically starts processing
+    messages when entered.
+    """
+
+    _response_streams: dict[RequestId, queue.Queue[JSONRPCResponse | JSONRPCError]]
+    _request_id: int
+    _in_flight: dict[RequestId, RequestResponder[ReceiveRequestT, SendResultT]]
+    _receive_request_type: type[ReceiveRequestT]
+    _receive_notification_type: type[ReceiveNotificationT]
+
+    def __init__(
+        self,
+        read_stream: queue.Queue,
+        write_stream: queue.Queue,
+        receive_request_type: type[ReceiveRequestT],
+        receive_notification_type: type[ReceiveNotificationT],
+        # If none, reading will never time out
+        read_timeout_seconds: timedelta | None = None,
+    ) -> None:
+        self._read_stream = read_stream
+        self._write_stream = write_stream
+        self._response_streams = {}
+        self._request_id = 0
+        self._receive_request_type = receive_request_type
+        self._receive_notification_type = receive_notification_type
+        self._session_read_timeout_seconds = read_timeout_seconds
+        self._in_flight = {}
+        self._exit_stack = ExitStack()
+
+    def __enter__(self) -> Self:
+        self._executor = ThreadPoolExecutor()
+        self._receiver_future = self._executor.submit(self._receive_loop)
+        return self
+
+    def check_receiver_status(self) -> None:
+        if self._receiver_future.done():
+            self._receiver_future.result()
+
+    def __exit__(
+        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
+    ) -> None:
+        self._exit_stack.close()
+        self._read_stream.put(None)
+        self._write_stream.put(None)
+
+    def send_request(
+        self,
+        request: SendRequestT,
+        result_type: type[ReceiveResultT],
+        request_read_timeout_seconds: timedelta | None = None,
+        metadata: MessageMetadata = None,
+    ) -> ReceiveResultT:
+        """
+        Sends a request and wait for a response. Raises an McpError if the
+        response contains an error. If a request read timeout is provided, it
+        will take precedence over the session read timeout.
+
+        Do not use this method to emit notifications! Use send_notification()
+        instead.
+        """
+        self.check_receiver_status()
+
+        request_id = self._request_id
+        self._request_id = request_id + 1
+
+        response_queue: queue.Queue[JSONRPCResponse | JSONRPCError] = queue.Queue()
+        self._response_streams[request_id] = response_queue
+
+        try:
+            jsonrpc_request = JSONRPCRequest(
+                jsonrpc="2.0",
+                id=request_id,
+                **request.model_dump(by_alias=True, mode="json", exclude_none=True),
+            )
+
+            self._write_stream.put(SessionMessage(message=JSONRPCMessage(jsonrpc_request), metadata=metadata))
+            timeout = DEFAULT_RESPONSE_READ_TIMEOUT
+            if request_read_timeout_seconds is not None:
+                timeout = float(request_read_timeout_seconds.total_seconds())
+            elif self._session_read_timeout_seconds is not None:
+                timeout = float(self._session_read_timeout_seconds.total_seconds())
+            while True:
+                try:
+                    response_or_error = response_queue.get(timeout=timeout)
+                    break
+                except queue.Empty:
+                    self.check_receiver_status()
+                    continue
+
+            if response_or_error is None:
+                raise MCPConnectionError(
+                    ErrorData(
+                        code=500,
+                        message="No response received",
+                    )
+                )
+            elif isinstance(response_or_error, JSONRPCError):
+                if response_or_error.error.code == 401:
+                    raise MCPAuthError(
+                        ErrorData(code=response_or_error.error.code, message=response_or_error.error.message)
+                    )
+                else:
+                    raise MCPConnectionError(
+                        ErrorData(code=response_or_error.error.code, message=response_or_error.error.message)
+                    )
+            else:
+                return result_type.model_validate(response_or_error.result)
+
+        finally:
+            self._response_streams.pop(request_id, None)
+
+    def send_notification(
+        self,
+        notification: SendNotificationT,
+        related_request_id: RequestId | None = None,
+    ) -> None:
+        """
+        Emits a notification, which is a one-way message that does not expect
+        a response.
+        """
+        self.check_receiver_status()
+
+        # Some transport implementations may need to set the related_request_id
+        # to attribute to the notifications to the request that triggered them.
+        jsonrpc_notification = JSONRPCNotification(
+            jsonrpc="2.0",
+            **notification.model_dump(by_alias=True, mode="json", exclude_none=True),
+        )
+        session_message = SessionMessage(
+            message=JSONRPCMessage(jsonrpc_notification),
+            metadata=ServerMessageMetadata(related_request_id=related_request_id) if related_request_id else None,
+        )
+        self._write_stream.put(session_message)
+
+    def _send_response(self, request_id: RequestId, response: SendResultT | ErrorData) -> None:
+        if isinstance(response, ErrorData):
+            jsonrpc_error = JSONRPCError(jsonrpc="2.0", id=request_id, error=response)
+            session_message = SessionMessage(message=JSONRPCMessage(jsonrpc_error))
+            self._write_stream.put(session_message)
+        else:
+            jsonrpc_response = JSONRPCResponse(
+                jsonrpc="2.0",
+                id=request_id,
+                result=response.model_dump(by_alias=True, mode="json", exclude_none=True),
+            )
+            session_message = SessionMessage(message=JSONRPCMessage(jsonrpc_response))
+            self._write_stream.put(session_message)
+
+    def _receive_loop(self) -> None:
+        """
+        Main message processing loop.
+        In a real synchronous implementation, this would likely run in a separate thread.
+        """
+        while True:
+            try:
+                # Attempt to receive a message (this would be blocking in a synchronous context)
+                message = self._read_stream.get(timeout=DEFAULT_RESPONSE_READ_TIMEOUT)
+                if message is None:
+                    break
+                if isinstance(message, HTTPStatusError):
+                    response_queue = self._response_streams.get(self._request_id - 1)
+                    if response_queue is not None:
+                        response_queue.put(
+                            JSONRPCError(
+                                jsonrpc="2.0",
+                                id=self._request_id - 1,
+                                error=ErrorData(code=message.response.status_code, message=message.args[0]),
+                            )
+                        )
+                    else:
+                        self._handle_incoming(RuntimeError(f"Received response with an unknown request ID: {message}"))
+                elif isinstance(message, Exception):
+                    self._handle_incoming(message)
+                elif isinstance(message.message.root, JSONRPCRequest):
+                    validated_request = self._receive_request_type.model_validate(
+                        message.message.root.model_dump(by_alias=True, mode="json", exclude_none=True)
+                    )
+
+                    responder = RequestResponder(
+                        request_id=message.message.root.id,
+                        request_meta=validated_request.root.params.meta if validated_request.root.params else None,
+                        request=validated_request,
+                        session=self,
+                        on_complete=lambda r: self._in_flight.pop(r.request_id, None),
+                    )
+
+                    self._in_flight[responder.request_id] = responder
+                    self._received_request(responder)
+
+                    if not responder._completed:
+                        self._handle_incoming(responder)
+
+                elif isinstance(message.message.root, JSONRPCNotification):
+                    try:
+                        notification = self._receive_notification_type.model_validate(
+                            message.message.root.model_dump(by_alias=True, mode="json", exclude_none=True)
+                        )
+                        # Handle cancellation notifications
+                        if isinstance(notification.root, CancelledNotification):
+                            cancelled_id = notification.root.params.requestId
+                            if cancelled_id in self._in_flight:
+                                self._in_flight[cancelled_id].cancel()
+                        else:
+                            self._received_notification(notification)
+                            self._handle_incoming(notification)
+                    except Exception as e:
+                        # For other validation errors, log and continue
+                        logging.warning(f"Failed to validate notification: {e}. Message was: {message.message.root}")
+                else:  # Response or error
+                    response_queue = self._response_streams.get(message.message.root.id)
+                    if response_queue is not None:
+                        response_queue.put(message.message.root)
+                    else:
+                        self._handle_incoming(RuntimeError(f"Server Error: {message}"))
+            except queue.Empty:
+                continue
+            except Exception as e:
+                logging.exception("Error in message processing loop")
+                raise
+
+    def _received_request(self, responder: RequestResponder[ReceiveRequestT, SendResultT]) -> None:
+        """
+        Can be overridden by subclasses to handle a request without needing to
+        listen on the message stream.
+
+        If the request is responded to within this method, it will not be
+        forwarded on to the message stream.
+        """
+        pass
+
+    def _received_notification(self, notification: ReceiveNotificationT) -> None:
+        """
+        Can be overridden by subclasses to handle a notification without needing
+        to listen on the message stream.
+        """
+        pass
+
+    def send_progress_notification(
+        self, progress_token: str | int, progress: float, total: float | None = None
+    ) -> None:
+        """
+        Sends a progress notification for a request that is currently being
+        processed.
+        """
+        pass
+
+    def _handle_incoming(
+        self,
+        req: RequestResponder[ReceiveRequestT, SendResultT] | ReceiveNotificationT | Exception,
+    ) -> None:
+        """A generic handler for incoming messages. Overwritten by subclasses."""
+        pass
@@ -0,0 +1,365 @@
+from datetime import timedelta
+from typing import Any, Protocol
+
+from pydantic import AnyUrl, TypeAdapter
+
+from configs import dify_config
+from core.mcp import types
+from core.mcp.entities import SUPPORTED_PROTOCOL_VERSIONS, RequestContext
+from core.mcp.session.base_session import BaseSession, RequestResponder
+
+DEFAULT_CLIENT_INFO = types.Implementation(name="Dify", version=dify_config.project.version)
+
+
+class SamplingFnT(Protocol):
+    def __call__(
+        self,
+        context: RequestContext["ClientSession", Any],
+        params: types.CreateMessageRequestParams,
+    ) -> types.CreateMessageResult | types.ErrorData: ...
+
+
+class ListRootsFnT(Protocol):
+    def __call__(self, context: RequestContext["ClientSession", Any]) -> types.ListRootsResult | types.ErrorData: ...
+
+
+class LoggingFnT(Protocol):
+    def __call__(
+        self,
+        params: types.LoggingMessageNotificationParams,
+    ) -> None: ...
+
+
+class MessageHandlerFnT(Protocol):
+    def __call__(
+        self,
+        message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
+    ) -> None: ...
+
+
+def _default_message_handler(
+    message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
+) -> None:
+    if isinstance(message, Exception):
+        raise ValueError(str(message))
+    elif isinstance(message, (types.ServerNotification | RequestResponder)):
+        pass
+
+
+def _default_sampling_callback(
+    context: RequestContext["ClientSession", Any],
+    params: types.CreateMessageRequestParams,
+) -> types.CreateMessageResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.INVALID_REQUEST,
+        message="Sampling not supported",
+    )
+
+
+def _default_list_roots_callback(
+    context: RequestContext["ClientSession", Any],
+) -> types.ListRootsResult | types.ErrorData:
+    return types.ErrorData(
+        code=types.INVALID_REQUEST,
+        message="List roots not supported",
+    )
+
+
+def _default_logging_callback(
+    params: types.LoggingMessageNotificationParams,
+) -> None:
+    pass
+
+
+ClientResponse: TypeAdapter[types.ClientResult | types.ErrorData] = TypeAdapter(types.ClientResult | types.ErrorData)
+
+
+class ClientSession(
+    BaseSession[
+        types.ClientRequest,
+        types.ClientNotification,
+        types.ClientResult,
+        types.ServerRequest,
+        types.ServerNotification,
+    ]
+):
+    def __init__(
+        self,
+        read_stream,
+        write_stream,
+        read_timeout_seconds: timedelta | None = None,
+        sampling_callback: SamplingFnT | None = None,
+        list_roots_callback: ListRootsFnT | None = None,
+        logging_callback: LoggingFnT | None = None,
+        message_handler: MessageHandlerFnT | None = None,
+        client_info: types.Implementation | None = None,
+    ) -> None:
+        super().__init__(
+            read_stream,
+            write_stream,
+            types.ServerRequest,
+            types.ServerNotification,
+            read_timeout_seconds=read_timeout_seconds,
+        )
+        self._client_info = client_info or DEFAULT_CLIENT_INFO
+        self._sampling_callback = sampling_callback or _default_sampling_callback
+        self._list_roots_callback = list_roots_callback or _default_list_roots_callback
+        self._logging_callback = logging_callback or _default_logging_callback
+        self._message_handler = message_handler or _default_message_handler
+
+    def initialize(self) -> types.InitializeResult:
+        sampling = types.SamplingCapability()
+        roots = types.RootsCapability(
+            # TODO: Should this be based on whether we
+            # _will_ send notifications, or only whether
+            # they're supported?
+            listChanged=True,
+        )
+
+        result = self.send_request(
+            types.ClientRequest(
+                types.InitializeRequest(
+                    method="initialize",
+                    params=types.InitializeRequestParams(
+                        protocolVersion=types.LATEST_PROTOCOL_VERSION,
+                        capabilities=types.ClientCapabilities(
+                            sampling=sampling,
+                            experimental=None,
+                            roots=roots,
+                        ),
+                        clientInfo=self._client_info,
+                    ),
+                )
+            ),
+            types.InitializeResult,
+        )
+
+        if result.protocolVersion not in SUPPORTED_PROTOCOL_VERSIONS:
+            raise RuntimeError(f"Unsupported protocol version from the server: {result.protocolVersion}")
+
+        self.send_notification(
+            types.ClientNotification(types.InitializedNotification(method="notifications/initialized"))
+        )
+
+        return result
+
+    def send_ping(self) -> types.EmptyResult:
+        """Send a ping request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.PingRequest(
+                    method="ping",
+                )
+            ),
+            types.EmptyResult,
+        )
+
+    def send_progress_notification(
+        self, progress_token: str | int, progress: float, total: float | None = None
+    ) -> None:
+        """Send a progress notification."""
+        self.send_notification(
+            types.ClientNotification(
+                types.ProgressNotification(
+                    method="notifications/progress",
+                    params=types.ProgressNotificationParams(
+                        progressToken=progress_token,
+                        progress=progress,
+                        total=total,
+                    ),
+                ),
+            )
+        )
+
+    def set_logging_level(self, level: types.LoggingLevel) -> types.EmptyResult:
+        """Send a logging/setLevel request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.SetLevelRequest(
+                    method="logging/setLevel",
+                    params=types.SetLevelRequestParams(level=level),
+                )
+            ),
+            types.EmptyResult,
+        )
+
+    def list_resources(self) -> types.ListResourcesResult:
+        """Send a resources/list request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.ListResourcesRequest(
+                    method="resources/list",
+                )
+            ),
+            types.ListResourcesResult,
+        )
+
+    def list_resource_templates(self) -> types.ListResourceTemplatesResult:
+        """Send a resources/templates/list request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.ListResourceTemplatesRequest(
+                    method="resources/templates/list",
+                )
+            ),
+            types.ListResourceTemplatesResult,
+        )
+
+    def read_resource(self, uri: AnyUrl) -> types.ReadResourceResult:
+        """Send a resources/read request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.ReadResourceRequest(
+                    method="resources/read",
+                    params=types.ReadResourceRequestParams(uri=uri),
+                )
+            ),
+            types.ReadResourceResult,
+        )
+
+    def subscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
+        """Send a resources/subscribe request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.SubscribeRequest(
+                    method="resources/subscribe",
+                    params=types.SubscribeRequestParams(uri=uri),
+                )
+            ),
+            types.EmptyResult,
+        )
+
+    def unsubscribe_resource(self, uri: AnyUrl) -> types.EmptyResult:
+        """Send a resources/unsubscribe request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.UnsubscribeRequest(
+                    method="resources/unsubscribe",
+                    params=types.UnsubscribeRequestParams(uri=uri),
+                )
+            ),
+            types.EmptyResult,
+        )
+
+    def call_tool(
+        self,
+        name: str,
+        arguments: dict[str, Any] | None = None,
+        read_timeout_seconds: timedelta | None = None,
+    ) -> types.CallToolResult:
+        """Send a tools/call request."""
+
+        return self.send_request(
+            types.ClientRequest(
+                types.CallToolRequest(
+                    method="tools/call",
+                    params=types.CallToolRequestParams(name=name, arguments=arguments),
+                )
+            ),
+            types.CallToolResult,
+            request_read_timeout_seconds=read_timeout_seconds,
+        )
+
+    def list_prompts(self) -> types.ListPromptsResult:
+        """Send a prompts/list request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.ListPromptsRequest(
+                    method="prompts/list",
+                )
+            ),
+            types.ListPromptsResult,
+        )
+
+    def get_prompt(self, name: str, arguments: dict[str, str] | None = None) -> types.GetPromptResult:
+        """Send a prompts/get request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.GetPromptRequest(
+                    method="prompts/get",
+                    params=types.GetPromptRequestParams(name=name, arguments=arguments),
+                )
+            ),
+            types.GetPromptResult,
+        )
+
+    def complete(
+        self,
+        ref: types.ResourceReference | types.PromptReference,
+        argument: dict[str, str],
+    ) -> types.CompleteResult:
+        """Send a completion/complete request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.CompleteRequest(
+                    method="completion/complete",
+                    params=types.CompleteRequestParams(
+                        ref=ref,
+                        argument=types.CompletionArgument(**argument),
+                    ),
+                )
+            ),
+            types.CompleteResult,
+        )
+
+    def list_tools(self) -> types.ListToolsResult:
+        """Send a tools/list request."""
+        return self.send_request(
+            types.ClientRequest(
+                types.ListToolsRequest(
+                    method="tools/list",
+                )
+            ),
+            types.ListToolsResult,
+        )
+
+    def send_roots_list_changed(self) -> None:
+        """Send a roots/list_changed notification."""
+        self.send_notification(
+            types.ClientNotification(
+                types.RootsListChangedNotification(
+                    method="notifications/roots/list_changed",
+                )
+            )
+        )
+
+    def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]) -> None:
+        ctx = RequestContext[ClientSession, Any](
+            request_id=responder.request_id,
+            meta=responder.request_meta,
+            session=self,
+            lifespan_context=None,
+        )
+
+        match responder.request.root:
+            case types.CreateMessageRequest(params=params):
+                with responder:
+                    response = self._sampling_callback(ctx, params)
+                    client_response = ClientResponse.validate_python(response)
+                    responder.respond(client_response)
+
+            case types.ListRootsRequest():
+                with responder:
+                    list_roots_response = self._list_roots_callback(ctx)
+                    client_response = ClientResponse.validate_python(list_roots_response)
+                    responder.respond(client_response)
+
+            case types.PingRequest():
+                with responder:
+                    return responder.respond(types.ClientResult(root=types.EmptyResult()))
+
+    def _handle_incoming(
+        self,
+        req: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
+    ) -> None:
+        """Handle incoming messages by forwarding to the message handler."""
+        self._message_handler(req)
+
+    def _received_notification(self, notification: types.ServerNotification) -> None:
+        """Handle notifications from the server."""
+        # Process specific notification types
+        match notification.root:
+            case types.LoggingMessageNotification(params=params):
+                self._logging_callback(params)
+            case _:
+                pass
@@ -0,0 +1,114 @@
+import json
+
+import httpx
+
+from configs import dify_config
+from core.mcp.types import ErrorData, JSONRPCError
+from core.model_runtime.utils.encoders import jsonable_encoder
+
+HTTP_REQUEST_NODE_SSL_VERIFY = dify_config.HTTP_REQUEST_NODE_SSL_VERIFY
+
+STATUS_FORCELIST = [429, 500, 502, 503, 504]
+
+
+def create_ssrf_proxy_mcp_http_client(
+    headers: dict[str, str] | None = None,
+    timeout: httpx.Timeout | None = None,
+) -> httpx.Client:
+    """Create an HTTPX client with SSRF proxy configuration for MCP connections.
+
+    Args:
+        headers: Optional headers to include in the client
+        timeout: Optional timeout configuration
+
+    Returns:
+        Configured httpx.Client with proxy settings
+    """
+    if dify_config.SSRF_PROXY_ALL_URL:
+        return httpx.Client(
+            verify=HTTP_REQUEST_NODE_SSL_VERIFY,
+            headers=headers or {},
+            timeout=timeout,
+            follow_redirects=True,
+            proxy=dify_config.SSRF_PROXY_ALL_URL,
+        )
+    elif dify_config.SSRF_PROXY_HTTP_URL and dify_config.SSRF_PROXY_HTTPS_URL:
+        proxy_mounts = {
+            "http://": httpx.HTTPTransport(proxy=dify_config.SSRF_PROXY_HTTP_URL, verify=HTTP_REQUEST_NODE_SSL_VERIFY),
+            "https://": httpx.HTTPTransport(
+                proxy=dify_config.SSRF_PROXY_HTTPS_URL, verify=HTTP_REQUEST_NODE_SSL_VERIFY
+            ),
+        }
+        return httpx.Client(
+            verify=HTTP_REQUEST_NODE_SSL_VERIFY,
+            headers=headers or {},
+            timeout=timeout,
+            follow_redirects=True,
+            mounts=proxy_mounts,
+        )
+    else:
+        return httpx.Client(
+            verify=HTTP_REQUEST_NODE_SSL_VERIFY,
+            headers=headers or {},
+            timeout=timeout,
+            follow_redirects=True,
+        )
+
+
+def ssrf_proxy_sse_connect(url, **kwargs):
+    """Connect to SSE endpoint with SSRF proxy protection.
+
+    This function creates an SSE connection using the configured proxy settings
+    to prevent SSRF attacks when connecting to external endpoints.
+
+    Args:
+        url: The SSE endpoint URL
+        **kwargs: Additional arguments passed to the SSE connection
+
+    Returns:
+        EventSource object for SSE streaming
+    """
+    from httpx_sse import connect_sse
+
+    # Extract client if provided, otherwise create one
+    client = kwargs.pop("client", None)
+    if client is None:
+        # Create client with SSRF proxy configuration
+        timeout = kwargs.pop(
+            "timeout",
+            httpx.Timeout(
+                timeout=dify_config.SSRF_DEFAULT_TIME_OUT,
+                connect=dify_config.SSRF_DEFAULT_CONNECT_TIME_OUT,
+                read=dify_config.SSRF_DEFAULT_READ_TIME_OUT,
+                write=dify_config.SSRF_DEFAULT_WRITE_TIME_OUT,
+            ),
+        )
+        headers = kwargs.pop("headers", {})
+        client = create_ssrf_proxy_mcp_http_client(headers=headers, timeout=timeout)
+        client_provided = False
+    else:
+        client_provided = True
+
+    # Extract method if provided, default to GET
+    method = kwargs.pop("method", "GET")
+
+    try:
+        return connect_sse(client, method, url, **kwargs)
+    except Exception:
+        # If we created the client, we need to clean it up on error
+        if not client_provided:
+            client.close()
+        raise
+
+
+def create_mcp_error_response(request_id: int | str | None, code: int, message: str, data=None):
+    """Create MCP error response"""
+    error_data = ErrorData(code=code, message=message, data=data)
+    json_response = JSONRPCError(
+        jsonrpc="2.0",
+        id=request_id or 1,
+        error=error_data,
+    )
+    json_data = json.dumps(jsonable_encoder(json_response))
+    sse_content = f"event: message\ndata: {json_data}\n\n".encode()
+    yield sse_content
@@ -1,7 +1,7 @@
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from decimal import Decimal
 from enum import StrEnum
-from typing import Optional
+from typing import Any, Optional

 from pydantic import BaseModel, Field

@@ -53,6 +53,37 @@ class LLMUsage(ModelUsage):
            latency=0.0,
        )

+    @classmethod
+    def from_metadata(cls, metadata: dict) -> "LLMUsage":
+        """
+        Create LLMUsage instance from metadata dictionary with default values.
+
+        Args:
+            metadata: Dictionary containing usage metadata
+
+        Returns:
+            LLMUsage instance with values from metadata or defaults
+        """
+        total_tokens = metadata.get("total_tokens", 0)
+        completion_tokens = metadata.get("completion_tokens", 0)
+        if total_tokens > 0 and completion_tokens == 0:
+            completion_tokens = total_tokens
+
+        return cls(
+            prompt_tokens=metadata.get("prompt_tokens", 0),
+            completion_tokens=completion_tokens,
+            total_tokens=total_tokens,
+            prompt_unit_price=Decimal(str(metadata.get("prompt_unit_price", 0))),
+            completion_unit_price=Decimal(str(metadata.get("completion_unit_price", 0))),
+            total_price=Decimal(str(metadata.get("total_price", 0))),
+            currency=metadata.get("currency", "USD"),
+            prompt_price_unit=Decimal(str(metadata.get("prompt_price_unit", 0))),
+            completion_price_unit=Decimal(str(metadata.get("completion_price_unit", 0))),
+            prompt_price=Decimal(str(metadata.get("prompt_price", 0))),
+            completion_price=Decimal(str(metadata.get("completion_price", 0))),
+            latency=metadata.get("latency", 0.0),
+        )
+
    def plus(self, other: "LLMUsage") -> "LLMUsage":
        """
        Add two LLMUsage instances together.
@@ -101,6 +132,20 @@ class LLMResult(BaseModel):
    system_fingerprint: Optional[str] = None


+class LLMStructuredOutput(BaseModel):
+    """
+    Model class for llm structured output.
+    """
+
+    structured_output: Optional[Mapping[str, Any]] = None
+
+
+class LLMResultWithStructuredOutput(LLMResult, LLMStructuredOutput):
+    """
+    Model class for llm result with structured output.
+    """
+
+
 class LLMResultChunkDelta(BaseModel):
    """
    Model class for llm result chunk delta.
@@ -123,6 +168,12 @@ class LLMResultChunk(BaseModel):
    delta: LLMResultChunkDelta


+class LLMResultChunkWithStructuredOutput(LLMResultChunk, LLMStructuredOutput):
+    """
+    Model class for llm result chunk with structured output.
+    """
+
+
 class NumTokensResult(PriceInfo):
    """
    Model class for number of tokens result.
@@ -0,0 +1,487 @@
+import json
+import logging
+from collections.abc import Sequence
+from typing import Optional
+from urllib.parse import urljoin
+
+from opentelemetry.trace import Status, StatusCode
+from sqlalchemy.orm import Session, sessionmaker
+
+from core.ops.aliyun_trace.data_exporter.traceclient import (
+    TraceClient,
+    convert_datetime_to_nanoseconds,
+    convert_to_span_id,
+    convert_to_trace_id,
+    generate_span_id,
+)
+from core.ops.aliyun_trace.entities.aliyun_trace_entity import SpanData
+from core.ops.aliyun_trace.entities.semconv import (
+    GEN_AI_COMPLETION,
+    GEN_AI_FRAMEWORK,
+    GEN_AI_MODEL_NAME,
+    GEN_AI_PROMPT,
+    GEN_AI_PROMPT_TEMPLATE_TEMPLATE,
+    GEN_AI_PROMPT_TEMPLATE_VARIABLE,
+    GEN_AI_RESPONSE_FINISH_REASON,
+    GEN_AI_SESSION_ID,
+    GEN_AI_SPAN_KIND,
+    GEN_AI_SYSTEM,
+    GEN_AI_USAGE_INPUT_TOKENS,
+    GEN_AI_USAGE_OUTPUT_TOKENS,
+    GEN_AI_USAGE_TOTAL_TOKENS,
+    GEN_AI_USER_ID,
+    INPUT_VALUE,
+    OUTPUT_VALUE,
+    RETRIEVAL_DOCUMENT,
+    RETRIEVAL_QUERY,
+    TOOL_DESCRIPTION,
+    TOOL_NAME,
+    TOOL_PARAMETERS,
+    GenAISpanKind,
+)
+from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import AliyunConfig
+from core.ops.entities.trace_entity import (
+    BaseTraceInfo,
+    DatasetRetrievalTraceInfo,
+    GenerateNameTraceInfo,
+    MessageTraceInfo,
+    ModerationTraceInfo,
+    SuggestedQuestionTraceInfo,
+    ToolTraceInfo,
+    WorkflowTraceInfo,
+)
+from core.rag.models.document import Document
+from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
+from core.workflow.entities.workflow_node_execution import (
+    WorkflowNodeExecution,
+    WorkflowNodeExecutionMetadataKey,
+    WorkflowNodeExecutionStatus,
+)
+from core.workflow.nodes import NodeType
+from models import Account, App, EndUser, TenantAccountJoin, WorkflowNodeExecutionTriggeredFrom, db
+
+logger = logging.getLogger(__name__)
+
+
+class AliyunDataTrace(BaseTraceInstance):
+    def __init__(
+        self,
+        aliyun_config: AliyunConfig,
+    ):
+        super().__init__(aliyun_config)
+        base_url = aliyun_config.endpoint.rstrip("/")
+        endpoint = urljoin(base_url, f"adapt_{aliyun_config.license_key}/api/otlp/traces")
+        self.trace_client = TraceClient(service_name=aliyun_config.app_name, endpoint=endpoint)
+
+    def trace(self, trace_info: BaseTraceInfo):
+        if isinstance(trace_info, WorkflowTraceInfo):
+            self.workflow_trace(trace_info)
+        if isinstance(trace_info, MessageTraceInfo):
+            self.message_trace(trace_info)
+        if isinstance(trace_info, ModerationTraceInfo):
+            pass
+        if isinstance(trace_info, SuggestedQuestionTraceInfo):
+            self.suggested_question_trace(trace_info)
+        if isinstance(trace_info, DatasetRetrievalTraceInfo):
+            self.dataset_retrieval_trace(trace_info)
+        if isinstance(trace_info, ToolTraceInfo):
+            self.tool_trace(trace_info)
+        if isinstance(trace_info, GenerateNameTraceInfo):
+            pass
+
+    def api_check(self):
+        return self.trace_client.api_check()
+
+    def get_project_url(self):
+        try:
+            return self.trace_client.get_project_url()
+        except Exception as e:
+            logger.info(f"Aliyun get run url failed: {str(e)}", exc_info=True)
+            raise ValueError(f"Aliyun get run url failed: {str(e)}")
+
+    def workflow_trace(self, trace_info: WorkflowTraceInfo):
+        trace_id = convert_to_trace_id(trace_info.workflow_run_id)
+        workflow_span_id = convert_to_span_id(trace_info.workflow_run_id, "workflow")
+        self.add_workflow_span(trace_id, workflow_span_id, trace_info)
+
+        workflow_node_executions = self.get_workflow_node_executions(trace_info)
+        for node_execution in workflow_node_executions:
+            node_span = self.build_workflow_node_span(node_execution, trace_id, trace_info, workflow_span_id)
+            self.trace_client.add_span(node_span)
+
+    def message_trace(self, trace_info: MessageTraceInfo):
+        message_data = trace_info.message_data
+        if message_data is None:
+            return
+        message_id = trace_info.message_id
+
+        user_id = message_data.from_account_id
+        if message_data.from_end_user_id:
+            end_user_data: Optional[EndUser] = (
+                db.session.query(EndUser).filter(EndUser.id == message_data.from_end_user_id).first()
+            )
+            if end_user_data is not None:
+                user_id = end_user_data.session_id
+
+        status: Status = Status(StatusCode.OK)
+        if trace_info.error:
+            status = Status(StatusCode.ERROR, trace_info.error)
+
+        trace_id = convert_to_trace_id(message_id)
+        message_span_id = convert_to_span_id(message_id, "message")
+        message_span = SpanData(
+            trace_id=trace_id,
+            parent_span_id=None,
+            span_id=message_span_id,
+            name="message",
+            start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
+            end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_USER_ID: str(user_id),
+                GEN_AI_SPAN_KIND: GenAISpanKind.CHAIN.value,
+                GEN_AI_FRAMEWORK: "dify",
+                INPUT_VALUE: json.dumps(trace_info.inputs, ensure_ascii=False),
+                OUTPUT_VALUE: str(trace_info.outputs),
+            },
+            status=status,
+        )
+        self.trace_client.add_span(message_span)
+
+        app_model_config = getattr(trace_info.message_data, "app_model_config", {})
+        pre_prompt = getattr(app_model_config, "pre_prompt", "")
+        inputs_data = getattr(trace_info.message_data, "inputs", {})
+        llm_span = SpanData(
+            trace_id=trace_id,
+            parent_span_id=message_span_id,
+            span_id=convert_to_span_id(message_id, "llm"),
+            name="llm",
+            start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
+            end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_USER_ID: str(user_id),
+                GEN_AI_SPAN_KIND: GenAISpanKind.LLM.value,
+                GEN_AI_FRAMEWORK: "dify",
+                GEN_AI_MODEL_NAME: trace_info.metadata.get("ls_model_name", ""),
+                GEN_AI_SYSTEM: trace_info.metadata.get("ls_provider", ""),
+                GEN_AI_USAGE_INPUT_TOKENS: str(trace_info.message_tokens),
+                GEN_AI_USAGE_OUTPUT_TOKENS: str(trace_info.answer_tokens),
+                GEN_AI_USAGE_TOTAL_TOKENS: str(trace_info.total_tokens),
+                GEN_AI_PROMPT_TEMPLATE_VARIABLE: json.dumps(inputs_data, ensure_ascii=False),
+                GEN_AI_PROMPT_TEMPLATE_TEMPLATE: pre_prompt,
+                GEN_AI_PROMPT: json.dumps(trace_info.inputs, ensure_ascii=False),
+                GEN_AI_COMPLETION: str(trace_info.outputs),
+                INPUT_VALUE: json.dumps(trace_info.inputs, ensure_ascii=False),
+                OUTPUT_VALUE: str(trace_info.outputs),
+            },
+            status=status,
+        )
+        self.trace_client.add_span(llm_span)
+
+    def dataset_retrieval_trace(self, trace_info: DatasetRetrievalTraceInfo):
+        if trace_info.message_data is None:
+            return
+        message_id = trace_info.message_id
+
+        documents_data = extract_retrieval_documents(trace_info.documents)
+        dataset_retrieval_span = SpanData(
+            trace_id=convert_to_trace_id(message_id),
+            parent_span_id=convert_to_span_id(message_id, "message"),
+            span_id=generate_span_id(),
+            name="dataset_retrieval",
+            start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
+            end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
+            attributes={
+                GEN_AI_SPAN_KIND: GenAISpanKind.RETRIEVER.value,
+                GEN_AI_FRAMEWORK: "dify",
+                RETRIEVAL_QUERY: str(trace_info.inputs),
+                RETRIEVAL_DOCUMENT: json.dumps(documents_data, ensure_ascii=False),
+                INPUT_VALUE: str(trace_info.inputs),
+                OUTPUT_VALUE: json.dumps(documents_data, ensure_ascii=False),
+            },
+        )
+        self.trace_client.add_span(dataset_retrieval_span)
+
+    def tool_trace(self, trace_info: ToolTraceInfo):
+        if trace_info.message_data is None:
+            return
+        message_id = trace_info.message_id
+
+        status: Status = Status(StatusCode.OK)
+        if trace_info.error:
+            status = Status(StatusCode.ERROR, trace_info.error)
+
+        tool_span = SpanData(
+            trace_id=convert_to_trace_id(message_id),
+            parent_span_id=convert_to_span_id(message_id, "message"),
+            span_id=generate_span_id(),
+            name=trace_info.tool_name,
+            start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
+            end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
+            attributes={
+                GEN_AI_SPAN_KIND: GenAISpanKind.TOOL.value,
+                GEN_AI_FRAMEWORK: "dify",
+                TOOL_NAME: trace_info.tool_name,
+                TOOL_DESCRIPTION: json.dumps(trace_info.tool_config, ensure_ascii=False),
+                TOOL_PARAMETERS: json.dumps(trace_info.tool_inputs, ensure_ascii=False),
+                INPUT_VALUE: json.dumps(trace_info.inputs, ensure_ascii=False),
+                OUTPUT_VALUE: str(trace_info.tool_outputs),
+            },
+            status=status,
+        )
+        self.trace_client.add_span(tool_span)
+
+    def get_workflow_node_executions(self, trace_info: WorkflowTraceInfo) -> Sequence[WorkflowNodeExecution]:
+        # through workflow_run_id get all_nodes_execution using repository
+        session_factory = sessionmaker(bind=db.engine)
+        # Find the app's creator account
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get the app to find its creator
+            app_id = trace_info.metadata.get("app_id")
+            if not app_id:
+                raise ValueError("No app_id found in trace_info metadata")
+
+            app = session.query(App).filter(App.id == app_id).first()
+            if not app:
+                raise ValueError(f"App with id {app_id} not found")
+
+            if not app.created_by:
+                raise ValueError(f"App with id {app_id} has no creator (created_by is None)")
+
+            service_account = session.query(Account).filter(Account.id == app.created_by).first()
+            if not service_account:
+                raise ValueError(f"Creator account with id {app.created_by} not found for app {app_id}")
+            current_tenant = (
+                session.query(TenantAccountJoin).filter_by(account_id=service_account.id, current=True).first()
+            )
+            if not current_tenant:
+                raise ValueError(f"Current tenant not found for account {service_account.id}")
+            service_account.set_tenant_id(current_tenant.tenant_id)
+        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
+            session_factory=session_factory,
+            user=service_account,
+            app_id=trace_info.metadata.get("app_id"),
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
+        )
+        # Get all executions for this workflow run
+        workflow_node_executions = workflow_node_execution_repository.get_by_workflow_run(
+            workflow_run_id=trace_info.workflow_run_id
+        )
+        return workflow_node_executions
+
+    def build_workflow_node_span(
+        self, node_execution: WorkflowNodeExecution, trace_id: int, trace_info: WorkflowTraceInfo, workflow_span_id: int
+    ):
+        try:
+            if node_execution.node_type == NodeType.LLM:
+                node_span = self.build_workflow_llm_span(trace_id, workflow_span_id, trace_info, node_execution)
+            elif node_execution.node_type == NodeType.KNOWLEDGE_RETRIEVAL:
+                node_span = self.build_workflow_retrieval_span(trace_id, workflow_span_id, trace_info, node_execution)
+            elif node_execution.node_type == NodeType.TOOL:
+                node_span = self.build_workflow_tool_span(trace_id, workflow_span_id, trace_info, node_execution)
+            else:
+                node_span = self.build_workflow_task_span(trace_id, workflow_span_id, trace_info, node_execution)
+            return node_span
+        except Exception:
+            return None
+
+    def get_workflow_node_status(self, node_execution: WorkflowNodeExecution) -> Status:
+        span_status: Status = Status(StatusCode.UNSET)
+        if node_execution.status == WorkflowNodeExecutionStatus.SUCCEEDED:
+            span_status = Status(StatusCode.OK)
+        elif node_execution.status in [WorkflowNodeExecutionStatus.FAILED, WorkflowNodeExecutionStatus.EXCEPTION]:
+            span_status = Status(StatusCode.ERROR, str(node_execution.error))
+        return span_status
+
+    def build_workflow_task_span(
+        self, trace_id: int, workflow_span_id: int, trace_info: WorkflowTraceInfo, node_execution: WorkflowNodeExecution
+    ) -> SpanData:
+        return SpanData(
+            trace_id=trace_id,
+            parent_span_id=workflow_span_id,
+            span_id=convert_to_span_id(node_execution.id, "node"),
+            name=node_execution.title,
+            start_time=convert_datetime_to_nanoseconds(node_execution.created_at),
+            end_time=convert_datetime_to_nanoseconds(node_execution.finished_at),
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_SPAN_KIND: GenAISpanKind.TASK.value,
+                GEN_AI_FRAMEWORK: "dify",
+                INPUT_VALUE: json.dumps(node_execution.inputs, ensure_ascii=False),
+                OUTPUT_VALUE: json.dumps(node_execution.outputs, ensure_ascii=False),
+            },
+            status=self.get_workflow_node_status(node_execution),
+        )
+
+    def build_workflow_tool_span(
+        self, trace_id: int, workflow_span_id: int, trace_info: WorkflowTraceInfo, node_execution: WorkflowNodeExecution
+    ) -> SpanData:
+        tool_des = {}
+        if node_execution.metadata:
+            tool_des = node_execution.metadata.get(WorkflowNodeExecutionMetadataKey.TOOL_INFO, {})
+        return SpanData(
+            trace_id=trace_id,
+            parent_span_id=workflow_span_id,
+            span_id=convert_to_span_id(node_execution.id, "node"),
+            name=node_execution.title,
+            start_time=convert_datetime_to_nanoseconds(node_execution.created_at),
+            end_time=convert_datetime_to_nanoseconds(node_execution.finished_at),
+            attributes={
+                GEN_AI_SPAN_KIND: GenAISpanKind.TOOL.value,
+                GEN_AI_FRAMEWORK: "dify",
+                TOOL_NAME: node_execution.title,
+                TOOL_DESCRIPTION: json.dumps(tool_des, ensure_ascii=False),
+                TOOL_PARAMETERS: json.dumps(node_execution.inputs if node_execution.inputs else {}, ensure_ascii=False),
+                INPUT_VALUE: json.dumps(node_execution.inputs if node_execution.inputs else {}, ensure_ascii=False),
+                OUTPUT_VALUE: json.dumps(node_execution.outputs, ensure_ascii=False),
+            },
+            status=self.get_workflow_node_status(node_execution),
+        )
+
+    def build_workflow_retrieval_span(
+        self, trace_id: int, workflow_span_id: int, trace_info: WorkflowTraceInfo, node_execution: WorkflowNodeExecution
+    ) -> SpanData:
+        input_value = ""
+        if node_execution.inputs:
+            input_value = str(node_execution.inputs.get("query", ""))
+        output_value = ""
+        if node_execution.outputs:
+            output_value = json.dumps(node_execution.outputs.get("result", []), ensure_ascii=False)
+        return SpanData(
+            trace_id=trace_id,
+            parent_span_id=workflow_span_id,
+            span_id=convert_to_span_id(node_execution.id, "node"),
+            name=node_execution.title,
+            start_time=convert_datetime_to_nanoseconds(node_execution.created_at),
+            end_time=convert_datetime_to_nanoseconds(node_execution.finished_at),
+            attributes={
+                GEN_AI_SPAN_KIND: GenAISpanKind.RETRIEVER.value,
+                GEN_AI_FRAMEWORK: "dify",
+                RETRIEVAL_QUERY: input_value,
+                RETRIEVAL_DOCUMENT: output_value,
+                INPUT_VALUE: input_value,
+                OUTPUT_VALUE: output_value,
+            },
+            status=self.get_workflow_node_status(node_execution),
+        )
+
+    def build_workflow_llm_span(
+        self, trace_id: int, workflow_span_id: int, trace_info: WorkflowTraceInfo, node_execution: WorkflowNodeExecution
+    ) -> SpanData:
+        process_data = node_execution.process_data or {}
+        outputs = node_execution.outputs or {}
+        usage_data = process_data.get("usage", {}) if "usage" in process_data else outputs.get("usage", {})
+        return SpanData(
+            trace_id=trace_id,
+            parent_span_id=workflow_span_id,
+            span_id=convert_to_span_id(node_execution.id, "node"),
+            name=node_execution.title,
+            start_time=convert_datetime_to_nanoseconds(node_execution.created_at),
+            end_time=convert_datetime_to_nanoseconds(node_execution.finished_at),
+            attributes={
+                GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                GEN_AI_SPAN_KIND: GenAISpanKind.LLM.value,
+                GEN_AI_FRAMEWORK: "dify",
+                GEN_AI_MODEL_NAME: process_data.get("model_name", ""),
+                GEN_AI_SYSTEM: process_data.get("model_provider", ""),
+                GEN_AI_USAGE_INPUT_TOKENS: str(usage_data.get("prompt_tokens", 0)),
+                GEN_AI_USAGE_OUTPUT_TOKENS: str(usage_data.get("completion_tokens", 0)),
+                GEN_AI_USAGE_TOTAL_TOKENS: str(usage_data.get("total_tokens", 0)),
+                GEN_AI_PROMPT: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
+                GEN_AI_COMPLETION: str(outputs.get("text", "")),
+                GEN_AI_RESPONSE_FINISH_REASON: outputs.get("finish_reason", ""),
+                INPUT_VALUE: json.dumps(process_data.get("prompts", []), ensure_ascii=False),
+                OUTPUT_VALUE: str(outputs.get("text", "")),
+            },
+            status=self.get_workflow_node_status(node_execution),
+        )
+
+    def add_workflow_span(self, trace_id: int, workflow_span_id: int, trace_info: WorkflowTraceInfo):
+        message_span_id = None
+        if trace_info.message_id:
+            message_span_id = convert_to_span_id(trace_info.message_id, "message")
+        user_id = trace_info.metadata.get("user_id")
+        status: Status = Status(StatusCode.OK)
+        if trace_info.error:
+            status = Status(StatusCode.ERROR, trace_info.error)
+        if message_span_id:  # chatflow
+            message_span = SpanData(
+                trace_id=trace_id,
+                parent_span_id=None,
+                span_id=message_span_id,
+                name="message",
+                start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
+                end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
+                attributes={
+                    GEN_AI_SESSION_ID: trace_info.metadata.get("conversation_id", ""),
+                    GEN_AI_USER_ID: str(user_id),
+                    GEN_AI_SPAN_KIND: GenAISpanKind.CHAIN.value,
+                    GEN_AI_FRAMEWORK: "dify",
+                    INPUT_VALUE: trace_info.workflow_run_inputs.get("sys.query", ""),
+                    OUTPUT_VALUE: json.dumps(trace_info.workflow_run_outputs, ensure_ascii=False),
+                },
+                status=status,
+            )
+            self.trace_client.add_span(message_span)
+
+        workflow_span = SpanData(
+            trace_id=trace_id,
+            parent_span_id=message_span_id,
+            span_id=workflow_span_id,
+            name="workflow",
+            start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
+            end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
+            attributes={
+                GEN_AI_USER_ID: str(user_id),
+                GEN_AI_SPAN_KIND: GenAISpanKind.CHAIN.value,
+                GEN_AI_FRAMEWORK: "dify",
+                INPUT_VALUE: json.dumps(trace_info.workflow_run_inputs, ensure_ascii=False),
+                OUTPUT_VALUE: json.dumps(trace_info.workflow_run_outputs, ensure_ascii=False),
+            },
+            status=status,
+        )
+        self.trace_client.add_span(workflow_span)
+
+    def suggested_question_trace(self, trace_info: SuggestedQuestionTraceInfo):
+        message_id = trace_info.message_id
+        status: Status = Status(StatusCode.OK)
+        if trace_info.error:
+            status = Status(StatusCode.ERROR, trace_info.error)
+        suggested_question_span = SpanData(
+            trace_id=convert_to_trace_id(message_id),
+            parent_span_id=convert_to_span_id(message_id, "message"),
+            span_id=convert_to_span_id(message_id, "suggested_question"),
+            name="suggested_question",
+            start_time=convert_datetime_to_nanoseconds(trace_info.start_time),
+            end_time=convert_datetime_to_nanoseconds(trace_info.end_time),
+            attributes={
+                GEN_AI_SPAN_KIND: GenAISpanKind.LLM.value,
+                GEN_AI_FRAMEWORK: "dify",
+                GEN_AI_MODEL_NAME: trace_info.metadata.get("ls_model_name", ""),
+                GEN_AI_SYSTEM: trace_info.metadata.get("ls_provider", ""),
+                GEN_AI_PROMPT: json.dumps(trace_info.inputs, ensure_ascii=False),
+                GEN_AI_COMPLETION: json.dumps(trace_info.suggested_question, ensure_ascii=False),
+                INPUT_VALUE: json.dumps(trace_info.inputs, ensure_ascii=False),
+                OUTPUT_VALUE: json.dumps(trace_info.suggested_question, ensure_ascii=False),
+            },
+            status=status,
+        )
+        self.trace_client.add_span(suggested_question_span)
+
+
+def extract_retrieval_documents(documents: list[Document]):
+    documents_data = []
+    for document in documents:
+        document_data = {
+            "content": document.page_content,
+            "metadata": {
+                "dataset_id": document.metadata.get("dataset_id"),
+                "doc_id": document.metadata.get("doc_id"),
+                "document_id": document.metadata.get("document_id"),
+            },
+            "score": document.metadata.get("score"),
+        }
+        documents_data.append(document_data)
+    return documents_data
@@ -0,0 +1,200 @@
+import hashlib
+import logging
+import random
+import socket
+import threading
+import uuid
+from collections import deque
+from collections.abc import Sequence
+from datetime import datetime
+from typing import Optional
+
+import requests
+from opentelemetry import trace as trace_api
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import ReadableSpan
+from opentelemetry.sdk.util.instrumentation import InstrumentationScope
+from opentelemetry.semconv.resource import ResourceAttributes
+
+from configs import dify_config
+from core.ops.aliyun_trace.entities.aliyun_trace_entity import SpanData
+
+INVALID_SPAN_ID = 0x0000000000000000
+INVALID_TRACE_ID = 0x00000000000000000000000000000000
+
+logger = logging.getLogger(__name__)
+
+
+class TraceClient:
+    def __init__(
+        self,
+        service_name: str,
+        endpoint: str,
+        max_queue_size: int = 1000,
+        schedule_delay_sec: int = 5,
+        max_export_batch_size: int = 50,
+    ):
+        self.endpoint = endpoint
+        self.resource = Resource(
+            attributes={
+                ResourceAttributes.SERVICE_NAME: service_name,
+                ResourceAttributes.SERVICE_VERSION: f"dify-{dify_config.project.version}-{dify_config.COMMIT_SHA}",
+                ResourceAttributes.DEPLOYMENT_ENVIRONMENT: f"{dify_config.DEPLOY_ENV}-{dify_config.EDITION}",
+                ResourceAttributes.HOST_NAME: socket.gethostname(),
+            }
+        )
+        self.span_builder = SpanBuilder(self.resource)
+        self.exporter = OTLPSpanExporter(endpoint=endpoint)
+
+        self.max_queue_size = max_queue_size
+        self.schedule_delay_sec = schedule_delay_sec
+        self.max_export_batch_size = max_export_batch_size
+
+        self.queue: deque = deque(maxlen=max_queue_size)
+        self.condition = threading.Condition(threading.Lock())
+        self.done = False
+
+        self.worker_thread = threading.Thread(target=self._worker, daemon=True)
+        self.worker_thread.start()
+
+        self._spans_dropped = False
+
+    def export(self, spans: Sequence[ReadableSpan]):
+        self.exporter.export(spans)
+
+    def api_check(self):
+        try:
+            response = requests.head(self.endpoint, timeout=5)
+            if response.status_code == 405:
+                return True
+            else:
+                logger.debug(f"AliyunTrace API check failed: Unexpected status code: {response.status_code}")
+                return False
+        except requests.exceptions.RequestException as e:
+            logger.debug(f"AliyunTrace API check failed: {str(e)}")
+            raise ValueError(f"AliyunTrace API check failed: {str(e)}")
+
+    def get_project_url(self):
+        return "https://arms.console.aliyun.com/#/llm"
+
+    def add_span(self, span_data: SpanData):
+        if span_data is None:
+            return
+        span: ReadableSpan = self.span_builder.build_span(span_data)
+        with self.condition:
+            if len(self.queue) == self.max_queue_size:
+                if not self._spans_dropped:
+                    logger.warning("Queue is full, likely spans will be dropped.")
+                    self._spans_dropped = True
+
+            self.queue.appendleft(span)
+            if len(self.queue) >= self.max_export_batch_size:
+                self.condition.notify()
+
+    def _worker(self):
+        while not self.done:
+            with self.condition:
+                if len(self.queue) < self.max_export_batch_size and not self.done:
+                    self.condition.wait(timeout=self.schedule_delay_sec)
+            self._export_batch()
+
+    def _export_batch(self):
+        spans_to_export: list[ReadableSpan] = []
+        with self.condition:
+            while len(spans_to_export) < self.max_export_batch_size and self.queue:
+                spans_to_export.append(self.queue.pop())
+
+        if spans_to_export:
+            try:
+                self.exporter.export(spans_to_export)
+            except Exception as e:
+                logger.debug(f"Error exporting spans: {e}")
+
+    def shutdown(self):
+        with self.condition:
+            self.done = True
+            self.condition.notify_all()
+        self.worker_thread.join()
+        self._export_batch()
+        self.exporter.shutdown()
+
+
+class SpanBuilder:
+    def __init__(self, resource):
+        self.resource = resource
+        self.instrumentation_scope = InstrumentationScope(
+            __name__,
+            "",
+            None,
+            None,
+        )
+
+    def build_span(self, span_data: SpanData) -> ReadableSpan:
+        span_context = trace_api.SpanContext(
+            trace_id=span_data.trace_id,
+            span_id=span_data.span_id,
+            is_remote=False,
+            trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.SAMPLED),
+            trace_state=None,
+        )
+
+        parent_span_context = None
+        if span_data.parent_span_id is not None:
+            parent_span_context = trace_api.SpanContext(
+                trace_id=span_data.trace_id,
+                span_id=span_data.parent_span_id,
+                is_remote=False,
+                trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.SAMPLED),
+                trace_state=None,
+            )
+
+        span = ReadableSpan(
+            name=span_data.name,
+            context=span_context,
+            parent=parent_span_context,
+            resource=self.resource,
+            attributes=span_data.attributes,
+            events=span_data.events,
+            links=span_data.links,
+            kind=trace_api.SpanKind.INTERNAL,
+            status=span_data.status,
+            start_time=span_data.start_time,
+            end_time=span_data.end_time,
+            instrumentation_scope=self.instrumentation_scope,
+        )
+        return span
+
+
+def generate_span_id() -> int:
+    span_id = random.getrandbits(64)
+    while span_id == INVALID_SPAN_ID:
+        span_id = random.getrandbits(64)
+    return span_id
+
+
+def convert_to_trace_id(uuid_v4: Optional[str]) -> int:
+    try:
+        uuid_obj = uuid.UUID(uuid_v4)
+        return uuid_obj.int
+    except Exception as e:
+        raise ValueError(f"Invalid UUID input: {e}")
+
+
+def convert_to_span_id(uuid_v4: Optional[str], span_type: str) -> int:
+    try:
+        uuid_obj = uuid.UUID(uuid_v4)
+    except Exception as e:
+        raise ValueError(f"Invalid UUID input: {e}")
+    combined_key = f"{uuid_obj.hex}-{span_type}"
+    hash_bytes = hashlib.sha256(combined_key.encode("utf-8")).digest()
+    span_id = int.from_bytes(hash_bytes[:8], byteorder="big", signed=False)
+    return span_id
+
+
+def convert_datetime_to_nanoseconds(start_time_a: Optional[datetime]) -> Optional[int]:
+    if start_time_a is None:
+        return None
+    timestamp_in_seconds = start_time_a.timestamp()
+    timestamp_in_nanoseconds = int(timestamp_in_seconds * 1e9)
+    return timestamp_in_nanoseconds
--- a/Show More
+++ b/Show More