diff --git a/apps/application/workflow/backend/__init__.py b/apps/application/workflow/backend/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/apps/application/workflow/backend/sandbox_shell.py b/apps/application/workflow/backend/sandbox_shell.py
new file mode 100644
index 00000000000..7dac1f90376
--- /dev/null
+++ b/apps/application/workflow/backend/sandbox_shell.py
@@ -0,0 +1,311 @@
+import getpass
+import os
+import re
+import shlex
+
+from deepagents.backends import LocalShellBackend
+from deepagents.backends.protocol import ExecuteResponse
+
+from common.utils.logger import maxkb_logger
+from maxkb.const import CONFIG
+
+_enable_sandbox = bool(int(CONFIG.get("SANDBOX", 1)))
+_run_user = "sandbox" if _enable_sandbox else getpass.getuser()
+_sandbox_python_sys_path = CONFIG.get_sandbox_python_package_paths().replace(",", ":")
+
+
+class SandboxShellBackend(LocalShellBackend):
+    def __init__(self, root_dir: str, **kwargs):
+        if "env" not in kwargs and not kwargs.get("inherit_env", False):
+            env = os.environ.copy()
+            python_path = env.get("PYTHONPATH", "")
+
+            # 将 sandbox Python 包路径分解为列表，检查每个路径是否已存在
+            existing_paths = set(python_path.split(os.pathsep))
+            sandbox_paths = _sandbox_python_sys_path.split(os.pathsep) if _sandbox_python_sys_path else []
+            new_paths = [p for p in sandbox_paths if p and p not in existing_paths]
+
+            if new_paths:
+                env["PYTHONPATH"] = (
+                    f"{os.pathsep.join(new_paths)}{os.pathsep}{python_path}"
+                    if python_path
+                    else os.pathsep.join(new_paths)
+                )
+
+            kwargs["env"] = env
+        super().__init__(root_dir=root_dir, **kwargs)
+
+    def _translate_virtual_paths(self, command: str) -> str:
+        """Translate virtual absolute paths in the command to real filesystem paths.
+
+        In virtual_mode=True, file tools (ls, glob, read_file) return virtual absolute
+        paths like /skills/foo.py which map to {root_dir}/skills/foo.py.  But execute()
+        runs a real shell where /skills/foo.py does not exist.  This method replaces
+        any path token that exists under root_dir with its real path, while leaving
+        genuine system paths (e.g. /usr/bin/python3) untouched.
+        """
+        root = str(self.cwd)
+
+        def translate(m: re.Match) -> str:
+            virtual_path = m.group(0)
+            real_path = root + virtual_path
+            return real_path if os.path.lexists(real_path) else virtual_path
+
+        # Match absolute-path-like tokens: / followed by a non-whitespace sequence
+        # that isn't clearly a flag (e.g. avoid matching -/something).
+        # Only translate when virtual_mode is active.
+        return re.sub(r'(?<![.\w\-])/[A-Za-z_][^\s\'"\\;|&><:,]*', translate, command)
+
+    def _consume_group(self, command: str, start_index: int) -> tuple[str, int]:
+        current = []
+        in_single_quote = False
+        in_double_quote = False
+        in_backticks = False
+        escaped = False
+        substitution_depth = 0
+        group_depth = 1
+        index = start_index + 1
+
+        while index < len(command):
+            char = command[index]
+
+            if escaped:
+                current.append(char)
+                escaped = False
+                index += 1
+                continue
+
+            if char == "\\" and not in_single_quote:
+                current.append(char)
+                escaped = True
+                index += 1
+                continue
+
+            if char == "`" and not in_single_quote:
+                in_backticks = not in_backticks
+                current.append(char)
+                index += 1
+                continue
+
+            if in_backticks:
+                current.append(char)
+                index += 1
+                continue
+
+            if char == "'" and not in_double_quote:
+                in_single_quote = not in_single_quote
+                current.append(char)
+                index += 1
+                continue
+
+            if char == '"' and not in_single_quote:
+                in_double_quote = not in_double_quote
+                current.append(char)
+                index += 1
+                continue
+
+            if in_single_quote or in_double_quote:
+                current.append(char)
+                index += 1
+                continue
+
+            if command.startswith("$(", index):
+                substitution_depth += 1
+                current.append("$(")
+                index += 2
+                continue
+
+            if substitution_depth:
+                if char == ")":
+                    substitution_depth -= 1
+                current.append(char)
+                index += 1
+                continue
+
+            if char == "(":
+                group_depth += 1
+                current.append(char)
+                index += 1
+                continue
+
+            if char == ")":
+                group_depth -= 1
+                if group_depth == 0:
+                    return "".join(current).strip(), index + 1
+                current.append(char)
+                index += 1
+                continue
+
+            current.append(char)
+            index += 1
+
+        raise ValueError("unclosed command group")
+
+    def _append_pending_command_part(self, parts: list[str | tuple[str, str]], current: list[str]) -> None:
+        part = "".join(current).strip()
+        if part:
+            parts.append(part)
+            return
+
+        if not parts:
+            parts.append("")
+            return
+
+        last_part = parts[-1]
+        if isinstance(last_part, str) and last_part in {";", "&&", "||", "|", "&"}:
+            parts.append("")
+
+    def _split_shell_command_list(self, command: str) -> list[str | tuple[str, str]]:
+        parts = []
+        current = []
+        in_single_quote = False
+        in_double_quote = False
+        in_backticks = False
+        escaped = False
+        substitution_depth = 0
+        index = 0
+
+        while index < len(command):
+            char = command[index]
+
+            if escaped:
+                current.append(char)
+                escaped = False
+                index += 1
+                continue
+
+            if char == "\\" and not in_single_quote:
+                current.append(char)
+                escaped = True
+                index += 1
+                continue
+
+            if char == "`" and not in_single_quote:
+                in_backticks = not in_backticks
+                current.append(char)
+                index += 1
+                continue
+
+            if in_backticks:
+                current.append(char)
+                index += 1
+                continue
+
+            if char == "'" and not in_double_quote:
+                in_single_quote = not in_single_quote
+                current.append(char)
+                index += 1
+                continue
+
+            if char == '"' and not in_single_quote:
+                in_double_quote = not in_double_quote
+                current.append(char)
+                index += 1
+                continue
+
+            if not in_single_quote and not in_double_quote:
+                if command.startswith("$(", index):
+                    substitution_depth += 1
+                    current.append("$(")
+                    index += 2
+                    continue
+
+                if substitution_depth:
+                    if char == ")":
+                        substitution_depth -= 1
+                    current.append(char)
+                    index += 1
+                    continue
+
+                if char == "(" and not "".join(current).strip():
+                    group_content, index = self._consume_group(command, index)
+                    parts.append(("group", group_content))
+                    current = []
+                    continue
+
+                if command.startswith("&&", index) or command.startswith("||", index):
+                    self._append_pending_command_part(parts, current)
+                    parts.append(command[index : index + 2])
+                    current = []
+                    index += 2
+                    continue
+
+                if char in {";", "|", "&"}:
+                    self._append_pending_command_part(parts, current)
+                    parts.append(char)
+                    current = []
+                    index += 1
+                    continue
+
+                if char == "\n":
+                    self._append_pending_command_part(parts, current)
+                    parts.append(";")
+                    current = []
+                    index += 1
+                    continue
+
+            current.append(char)
+            index += 1
+
+        self._append_pending_command_part(parts, current)
+        return parts
+
+    def _build_sandbox_command(self, command: str) -> str:
+        prefix = (
+            "env -i LD_PRELOAD=/opt/maxkb-app/sandbox/lib/sandbox.so "
+            f'PATH="${{PATH}}" PYTHONPATH="${{PYTHONPATH}}" gosu {_run_user} '
+        )
+        parts = self._split_shell_command_list(command)
+        sandboxed_parts = []
+        expect_command = True
+
+        for part in parts:
+            if expect_command:
+                if isinstance(part, tuple):
+                    group_kind, group_content = part
+                    if group_kind != "group":
+                        raise ValueError(f"unsupported command part: {group_kind}")
+                    if not group_content:
+                        raise ValueError("empty command group")
+                    sandboxed_parts.append(f"( {self._build_sandbox_command(group_content)} )")
+                elif not part:
+                    raise ValueError("empty command")
+                else:
+                    tokens = shlex.split(part)
+                    if not tokens:
+                        raise ValueError("empty command")
+                    sandboxed_parts.append(prefix + " ".join(shlex.quote(token) for token in tokens))
+            else:
+                if part not in {";", "&&", "||", "|", "&"}:
+                    raise ValueError(f"unsupported shell operator: {part}")
+                sandboxed_parts.append(part)
+
+            expect_command = not expect_command
+
+        if expect_command:
+            raise ValueError("command cannot end with a shell operator")
+
+        return " ".join(sandboxed_parts)
+
+    def execute(
+        self,
+        command: str,
+        *,
+        timeout: int | None = None,
+    ) -> ExecuteResponse:
+        if self.virtual_mode:
+            command = self._translate_virtual_paths(command)
+
+        if _enable_sandbox:
+            # 用 runuser 在子进程里切换用户，父进程凭据保持不变，
+            # 避免父进程 ruid/euid 不一致导致 execve 报 Permission denied
+            try:
+                # 将命令列表拆成多个简单命令，并分别在 sandbox 用户下执行。
+                # 每个简单命令仍按 argv 重新 quote，避免 $()、反引号等在父 shell 中展开。
+                command = self._build_sandbox_command(command)
+            except ValueError as e:
+                return ExecuteResponse(output=f"Invalid command: {e}", exit_code=1)
+            # command = f"runuser -u {_run_user} -- env -i PATH=${{PATH}} {command}"
+
+        maxkb_logger.debug(f"Executing command in sandbox: {command}")
+        return super().execute(command=command, timeout=timeout)
diff --git a/apps/application/workflow/nodes/ai_chat_node/agent.py b/apps/application/workflow/nodes/ai_chat_node/agent.py
new file mode 100644
index 00000000000..ea322db9d5f
--- /dev/null
+++ b/apps/application/workflow/nodes/ai_chat_node/agent.py
@@ -0,0 +1,743 @@
+# coding=utf-8
+"""
+@project: MaxKB
+@Author:虎虎虎
+@file:  agent.py
+@date:  2026/9/14 16:59
+@desc:  AI 对话节点的 Agent（MCP / deepagents）执行逻辑。
+
+从 application/flow/tools.py 抽离，供新工作流引擎的 ai_chat_node 使用，
+避免新引擎反向依赖旧引擎的 flow.tools 模块。
+"""
+
+import asyncio
+import io
+import json
+import os
+import queue
+import re
+import shutil
+import threading
+import time
+import zipfile
+
+import langchain_core.messages.ai as _lc_ai_module
+import uuid_utils.compat as uuid
+from asgiref.sync import sync_to_async
+from deepagents import create_deep_agent
+from django.db.models import OuterRef, QuerySet, Subquery
+from langchain_core.messages import AIMessageChunk, ToolMessage
+from langchain_core.tools import StructuredTool
+from langchain_core.utils._merge import merge_lists as _original_merge_lists
+from langchain_mcp_adapters.client import MultiServerMCPClient
+from langgraph.checkpoint.memory import MemorySaver
+from pydantic import Field, create_model
+
+from application.workflow.backend.sandbox_shell import SandboxShellBackend
+from application.workflow.message.aggregator import AggregationManager
+from application.workflow.status import Status
+from common.utils.logger import maxkb_logger
+from knowledge.models import File
+from knowledge.models.knowledge_action import State
+from maxkb.const import CONFIG
+from tools.models import Tool, ToolRecord, ToolType, ToolWorkflowVersion
+
+
+# ---------------------------------------------------------------------------
+# Fix: qwen's OpenAI-compatible streaming sends id='' (empty string) for
+# intermediate tool_call_chunks while only the first chunk carries the real
+# id ('call_xxx...'). langchain-core's merge_lists treats '' != 'call_xxx' as
+# an ID conflict and _appends_ instead of merging → the accumulated AIMessage
+# ends up with two separate tool_calls (one with empty args, one with empty
+# id) instead of one correct entry. This causes the Qwen API to reject the
+# next request with "function.arguments must be in JSON format".
+#
+# Patch: normalise id='' → None for items that have an 'index' key
+# (i.e. tool_call_chunk dicts). merge_lists treats None as "no id" and will
+# merge with any existing entry, keeping the real id from the first chunk.
+# ---------------------------------------------------------------------------
+def _merge_lists_normalize_empty_tool_chunk_ids(left, *others):
+    """Wrapper around merge_lists that normalises empty-string IDs to None in
+    tool_call_chunk items (those with an 'index' key) so that qwen streaming
+    chunks with id='' are merged correctly by index."""
+
+    def _norm(lst):
+        if lst is None:
+            return lst
+        result = []
+        for item in lst:
+            if isinstance(item, dict) and "index" in item and item.get("id") == "":
+                item = {**item, "id": None}
+            result.append(item)
+        return result
+
+    return _original_merge_lists(
+        _norm(left),
+        *[_norm(o) for o in others],
+    )
+
+
+# Replace the module-level reference used by add_ai_message_chunks in ai.py
+_lc_ai_module.merge_lists = _merge_lists_normalize_empty_tool_chunk_ids
+
+
+def generate_tool_message_complete(icon, name, input_content, output_content):
+    """生成包含输入和输出的工具消息模版"""
+    # 确保输入内容是字符串，如果不是则尝试转换为 JSON 字符串
+    if not isinstance(input_content, str):
+        input_content = json.dumps(input_content, ensure_ascii=False)
+    # 格式化输出
+    if not isinstance(output_content, str):
+        output_content = json.dumps(output_content, ensure_ascii=False)
+    content = {
+        "icon": icon,
+        "title": name,
+        "type": "simple-tool-calls",
+        "content": {"input": input_content, "output": output_content},
+    }
+    return f"<tool_calls_render>{json.dumps(content, ensure_ascii=False)}</tool_calls_render>"
+
+
+# 全局单例事件循环
+_global_loop = None
+_loop_thread = None
+_loop_lock = threading.Lock()
+
+
+def get_global_loop():
+    """获取全局共享的事件循环"""
+    global _global_loop, _loop_thread
+
+    with _loop_lock:
+        if _global_loop is None:
+            _global_loop = asyncio.new_event_loop()
+
+            def run_forever():
+                asyncio.set_event_loop(_global_loop)
+                _global_loop.run_forever()
+
+            _loop_thread = threading.Thread(target=run_forever, daemon=True, name="GlobalAsyncLoop")
+            _loop_thread.start()
+
+    return _global_loop
+
+
+def _extract_tool_id(raw_id):
+    """从 raw_id 中提取最后一个符合 call_... 模式的 id，若无匹配则返回原值或 None"""
+    if not raw_id:
+        return None
+    if not isinstance(raw_id, str):
+        raw_id = str(raw_id)
+
+    s = raw_id
+    prefix = "call_"
+    positions = [m.start() for m in re.finditer(re.escape(prefix), s)]
+    if not positions:
+        return raw_id
+
+    # 取最后一个前缀位置，截到下一个前缀或结尾
+    start = positions[-1]
+    end = len(s)
+    for pos in positions:
+        if pos > start:
+            end = pos
+            break
+
+    tool_id = s[start:end]
+    return tool_id or raw_id
+
+
+async def _initialize_skills(mcp_servers, temp_dir):
+    skills_dir = os.path.join(temp_dir, "skills")
+    mcp_config = json.loads(mcp_servers)
+    if "skills" in mcp_config:
+        skill_file_items = mcp_config.pop("skills")
+        for skill_file in skill_file_items:
+            # 使用 sync_to_async 包装 ORM 查询
+            file = await sync_to_async(lambda: QuerySet(File).filter(id=skill_file["file_id"]).first())()
+            if not file:
+                continue
+            # get_bytes 可能也涉及 IO，也用 sync_to_async 包装
+            file_bytes = await sync_to_async(file.get_bytes)()
+            params = skill_file.get("params", {})
+            with zipfile.ZipFile(io.BytesIO(file_bytes), "r") as zip_ref:
+                members = [m for m in zip_ref.namelist() if not m.startswith("__MACOSX/") and "__MACOSX" not in m]
+                for member in members:
+                    if ".." in member or member.startswith("/"):
+                        raise ValueError(f"非法路径: {member}")
+                zip_ref.extractall(skills_dir, members=members)
+
+                # 获取技能解压后的顶级目录名
+                top_level_dirs = set()
+                for member in members:
+                    parts = member.split("/")
+                    if parts[0]:
+                        top_level_dirs.add(parts[0])
+
+                # 将 params 写入每个顶级目录下的 .env 文件
+                if params:
+                    env_lines = []
+                    for key, value in params.items():
+                        # 对含空格或特殊字符的值加引号
+                        env_lines.append(f"{key}={value}")
+                    env_content = "\n".join(env_lines) + "\n"
+                    for top_dir in top_level_dirs:
+                        env_path = os.path.join(skills_dir, top_dir, ".env")
+                        with open(env_path, "w", encoding="utf-8") as f:
+                            f.write(env_content)
+
+        os.system("chmod -R g+rx " + temp_dir)  # 确保技能目录可访问
+
+    client = MultiServerMCPClient(mcp_config)
+
+    return client
+
+
+async def _yield_mcp_response(
+    chat_model,
+    system_prompt,
+    message_list,
+    mcp_servers,
+    mcp_output_enable=True,
+    tool_init_params={},
+    source_id=None,
+    source_type=None,
+    temp_dir=None,
+    chat_id=None,
+    extra_tools=None,
+):
+    try:
+        checkpointer = MemorySaver()
+        client = await _initialize_skills(mcp_servers, temp_dir)
+        tools = await client.get_tools()
+        for tool in tools:
+            tool.handle_tool_error = True
+        if extra_tools:
+            for tool in extra_tools:
+                tools.append(tool)
+
+        agent = create_deep_agent(
+            model=chat_model,
+            backend=SandboxShellBackend(root_dir=temp_dir, virtual_mode=True),
+            skills=["/skills"],
+            tools=tools,
+            system_prompt=system_prompt,
+            interrupt_on={"write_file": False, "read_file": False, "edit_file": False},
+            checkpointer=checkpointer,
+        )
+        recursion_limit = int(CONFIG.get("LANGCHAIN_GRAPH_RECURSION_LIMIT", "100"))
+        response = agent.astream(
+            {"messages": message_list},
+            config={"recursion_limit": recursion_limit, "configurable": {"thread_id": chat_id}},
+            stream_mode="messages",
+        )
+
+        tool_calls_info = {}  # tool_id -> {'name': ..., 'input': ...}
+        # key(index/id) -> {'id': ..., 'name': ..., 'arguments': ...}
+        _tool_fragments = {}
+
+        def _merge_arguments(entry, part_args):
+            if not isinstance(part_args, str):
+                try:
+                    part_args = json.dumps(part_args, ensure_ascii=False)
+                except Exception:
+                    part_args = str(part_args) if part_args else ""
+            if not part_args:
+                return
+
+            # Some providers first emit placeholder args like "{}" and then
+            # stream the real JSON fragments via later chunks. Prefer fragments.
+            if entry["arguments"] in ("{}", "[]") and part_args.startswith("{"):
+                entry["arguments"] = part_args
+                return
+
+            if entry["arguments"]:
+                try:
+                    existing_obj = json.loads(entry["arguments"])
+                    new_obj = json.loads(part_args)
+                    if isinstance(existing_obj, dict) and isinstance(new_obj, dict):
+                        merged = {**existing_obj, **new_obj}
+                        entry["arguments"] = json.dumps(merged, ensure_ascii=False)
+                    else:
+                        entry["arguments"] += part_args
+                except (json.JSONDecodeError, ValueError):
+                    entry["arguments"] += part_args
+            else:
+                entry["arguments"] = part_args
+
+        def _get_fragment_key(idx, raw_id):
+            if idx is not None:
+                return f"idx:{idx}"
+            if raw_id and str(raw_id).strip():
+                return f"id:{_extract_tool_id(str(raw_id).strip())}"
+            return None
+
+        def _upsert_fragment(key, raw_id, func_name, part_args):
+            if key is None:
+                return
+            entry = _tool_fragments.setdefault(key, {"id": "", "name": "", "arguments": ""})
+
+            if raw_id and str(raw_id).strip():
+                new_id = str(raw_id).strip()
+                if entry.get("completed") and entry.get("id") and entry["id"] != new_id:
+                    maxkb_logger.debug(f"Resetting completed fragment {key}: old ID {entry['id']} -> new ID {new_id}")
+                    entry.clear()
+                    entry.update({"id": "", "name": "", "arguments": ""})
+                entry["id"] = new_id
+
+            if func_name:
+                entry["name"] = func_name
+
+            _merge_arguments(entry, part_args)
+
+        async for chunk in response:
+            # print(chunk)
+            if isinstance(chunk[0], AIMessageChunk):
+                # ----------------------------------------------------------------
+                # 1. 从 tool_call_chunks 中聚合工具调用片段
+                #    (qwen/OpenAI streaming 通过 tool_call_chunks 传递，
+                #     additional_kwargs['tool_calls'] 在流式时通常为空)
+                # ----------------------------------------------------------------
+                for tc_chunk in chunk[0].tool_call_chunks or []:
+                    raw_id = tc_chunk.get("id")
+                    key = _get_fragment_key(tc_chunk.get("index"), raw_id)
+                    _upsert_fragment(key, raw_id, tc_chunk.get("name"), tc_chunk.get("args", ""))
+
+                # ----------------------------------------------------------------
+                # 1.1 兼容部分模型将工具调用放在 chunk.tool_calls，且 tool_call_chunks
+                #     的 index 为空（例如 ollama/qwen）
+                # ----------------------------------------------------------------
+                has_tool_call_chunks = bool(chunk[0].tool_call_chunks)
+                for tool_call in chunk[0].tool_calls or []:
+                    raw_id = tool_call.get("id")
+                    part_args = tool_call.get("args", "")
+                    # qwen-plus often emits {} here as a placeholder while
+                    # the real args are split in tool_call_chunks/invalid_tool_calls.
+                    if has_tool_call_chunks and (part_args == "" or part_args == {} or part_args == []):
+                        part_args = ""
+                    key = _get_fragment_key(tool_call.get("index"), raw_id)
+                    _upsert_fragment(key, raw_id, tool_call.get("name"), part_args)
+
+                # ----------------------------------------------------------------
+                # 1.2 兼容 invalid_tool_calls 分片（部分模型会把中间 JSON 片段放这里）
+                # ----------------------------------------------------------------
+                for invalid_tool_call in chunk[0].invalid_tool_calls or []:
+                    raw_id = invalid_tool_call.get("id")
+                    key = _get_fragment_key(invalid_tool_call.get("index"), raw_id)
+                    _upsert_fragment(key, raw_id, invalid_tool_call.get("name"), invalid_tool_call.get("args", ""))
+
+                # ----------------------------------------------------------------
+                # 2. 兼容 additional_kwargs['tool_calls'] 方式（旧格式/非流式情况）
+                # ----------------------------------------------------------------
+                legacy_tool_calls = chunk[0].additional_kwargs.get("tool_calls", [])
+                for tool_call in legacy_tool_calls:
+                    raw_id = tool_call.get("id")
+                    func = tool_call.get("function", {})
+                    if isinstance(func, dict):
+                        func_name = func.get("name")
+                        part_args = func.get("arguments", "")
+                    else:
+                        func_name = tool_call.get("name")
+                        part_args = tool_call.get("arguments", "")
+                    key = _get_fragment_key(tool_call.get("index"), raw_id)
+                    _upsert_fragment(key, raw_id, func_name, part_args)
+
+                # ----------------------------------------------------------------
+                # 3. 检测工具调用结束，更新 tool_calls_info
+                # ----------------------------------------------------------------
+                is_finish_chunk = (
+                    chunk[0].response_metadata.get("finish_reason") == "tool_calls" or chunk[0].chunk_position == "last"
+                )
+
+                if is_finish_chunk:
+                    # 在 finish chunk 时，将所有未完成的 fragment 标记完成并更新 tool_calls_info
+                    maxkb_logger.debug(f"Processing finish chunk. Tool fragments: {_tool_fragments}")
+                    for idx, entry in _tool_fragments.items():
+                        if entry.get("completed"):
+                            maxkb_logger.debug(f"Skipping fragment {idx}: already completed")
+                            continue
+                        if not entry.get("id"):
+                            maxkb_logger.debug(f"Skipping fragment {idx}: missing id. Fragment: {entry}")
+                            continue
+                        if not entry.get("arguments"):
+                            maxkb_logger.debug(f"Skipping fragment {idx}: missing arguments. Fragment: {entry}")
+                            continue
+
+                        if not entry.get("completed") and entry.get("id") and entry.get("arguments"):
+                            try:
+                                parsed_args = json.loads(entry["arguments"])
+                                filtered_args = (
+                                    {k: v for k, v in parsed_args.items() if k not in tool_init_params}
+                                    if tool_init_params
+                                    else parsed_args
+                                )
+                                normalized_id = _extract_tool_id(entry["id"])
+                                info = {"name": entry["name"], "input": json.dumps(filtered_args, ensure_ascii=False)}
+                                tool_calls_info[entry["id"]] = info
+                                if normalized_id and normalized_id != entry["id"]:
+                                    tool_calls_info[normalized_id] = info
+                                entry["completed"] = True
+                                maxkb_logger.debug(f"Added tool call {entry['id']} to tool_calls_info")
+                            except (json.JSONDecodeError, ValueError) as e:
+                                # JSON parsing failed, but still add to tool_calls_info with raw arguments
+                                # to prevent "Tool ID not found" errors when ToolMessage arrives
+                                maxkb_logger.warning(
+                                    f"Failed to parse tool arguments at finish for tool {entry.get('id', 'unknown')}: "
+                                    f"{entry['arguments']}, error: {e}. Using raw arguments."
+                                )
+                                normalized_id = _extract_tool_id(entry["id"])
+                                info = {
+                                    "name": entry["name"],
+                                    # Use raw arguments
+                                    "input": entry["arguments"],
+                                }
+                                tool_calls_info[entry["id"]] = info
+                                if normalized_id and normalized_id != entry["id"]:
+                                    tool_calls_info[normalized_id] = info
+                                entry["completed"] = True
+
+                # ----------------------------------------------------------------
+                # 4. 修复 tool_call_chunks 中的空 id（回填已知 id）
+                # ----------------------------------------------------------------
+                if chunk[0].tool_call_chunks:
+                    for tc_chunk in chunk[0].tool_call_chunks:
+                        key = _get_fragment_key(tc_chunk.get("index"), tc_chunk.get("id"))
+                        if key is not None:
+                            frag = _tool_fragments.get(key)
+                            if frag and frag.get("id") and not tc_chunk.get("id"):
+                                tc_chunk["id"] = frag["id"]
+
+                # ----------------------------------------------------------------
+                # 5. 修复 additional_kwargs['tool_calls']（兼容旧格式）
+                #    仅在 finish chunk 时写入完整参数，避免污染中间 chunk 的
+                #    additional_kwargs（中间 chunk 会被 ainvoke 累积，如果写入
+                #    不完整 JSON 会导致下一轮 API 调用出现 arguments 非 JSON 格式错误）
+                # ----------------------------------------------------------------
+                if legacy_tool_calls and is_finish_chunk:
+                    fixed_tool_calls = []
+                    for tool_call in legacy_tool_calls:
+                        key = _get_fragment_key(tool_call.get("index"), tool_call.get("id"))
+                        frag = _tool_fragments.get(key) if key is not None else None
+                        tc = dict(tool_call)
+                        if frag and frag.get("id") and not tc.get("id"):
+                            tc["id"] = frag["id"]
+                        if frag and isinstance(tc.get("function"), dict):
+                            tc["function"] = dict(tc["function"])
+                            if frag.get("completed"):
+                                tc["function"]["arguments"] = frag["arguments"]
+                        fixed_tool_calls.append(tc)
+                    chunk[0].additional_kwargs["tool_calls"] = fixed_tool_calls
+
+                yield chunk[0]
+
+            if mcp_output_enable and isinstance(chunk[0], ToolMessage):
+                tool_id = chunk[0].tool_call_id
+                normalized_tool_id = _extract_tool_id(tool_id)
+                tool_info = tool_calls_info.get(tool_id) or tool_calls_info.get(normalized_tool_id)
+
+                if tool_info:
+                    try:
+                        if isinstance(chunk[0].content, str):
+                            tool_result = json.loads(chunk[0].content)
+                        elif isinstance(chunk[0].content, dict):
+                            tool_result = chunk[0].content
+                        elif isinstance(chunk[0].content, list):
+                            tool_result = chunk[0].content[0] if len(chunk[0].content) > 0 else {}
+                        else:
+                            tool_result = {}
+                        text = tool_result.get("text") if "text" in tool_result else None
+                        text_result = json.loads(text) if text else tool_result
+                        if text:
+                            tool_lib_id = text_result.pop("tool_id") if "tool_id" in text_result else None
+                        else:
+                            tool_lib_id = tool_result.pop("tool_id") if "tool_id" in tool_result else None
+                        if tool_lib_id:
+                            await save_tool_record(tool_lib_id, tool_info, tool_result, source_id, source_type)
+                        tool_result = json.dumps(text_result, ensure_ascii=False)
+                    except Exception:
+                        tool_result = chunk[0].content
+                    content = generate_tool_message_complete(
+                        tool_info.get("icon", ""), tool_info["name"], tool_info["input"], tool_result
+                    )
+                    chunk[0].content = content
+                else:
+                    maxkb_logger.warning(
+                        f"Tool ID {tool_id} not found in tool_calls_info. "
+                        f"Normalized Tool ID: {normalized_tool_id}. "
+                        f"Available IDs: {list(tool_calls_info.keys())}. "
+                        f"Tool fragments at this point: {_tool_fragments}"
+                    )
+
+                yield chunk[0]
+
+    except ExceptionGroup as eg:
+
+        def get_real_error(exc):
+            if isinstance(exc, ExceptionGroup):
+                return get_real_error(exc.exceptions[0])
+            return exc
+
+        real_error = get_real_error(eg)
+        error_msg = f"{type(real_error).__name__}: {str(real_error)}"
+        raise RuntimeError(error_msg) from None
+
+    except Exception as e:
+        error_msg = f"{type(e).__name__}: {str(e)}"
+        raise RuntimeError(error_msg) from None
+
+
+async def save_tool_record(tool_id, tool_info, tool_result, source_id, source_type):
+    tool = await sync_to_async(lambda: QuerySet(Tool).filter(id=tool_id).first())()
+    tool_info["icon"] = tool.icon
+    tool_record = ToolRecord(
+        id=uuid.uuid7(),
+        workspace_id=tool.workspace_id,
+        tool_id=tool_id,
+        source_type=source_type,
+        source_id=source_id,
+        meta={"input": tool_info["input"], "output": tool_result},
+        state=State.SUCCESS,
+    )
+    await sync_to_async(tool_record.save)()
+
+
+def mcp_response_generator(
+    chat_model,
+    system_prompt,
+    message_list,
+    mcp_servers,
+    mcp_output_enable=True,
+    tool_init_params={},
+    source_id=None,
+    source_type=None,
+    chat_id=None,
+    extra_tools=None,
+):
+    """使用全局事件循环，不创建新实例"""
+    result_queue = queue.Queue()
+    loop = get_global_loop()  # 使用共享循环
+    # 创建临时文件夹
+    if chat_id:
+        temp_dir = os.path.join("/tmp", chat_id)
+    else:
+        temp_dir = os.path.join("/tmp", str(uuid.uuid7()))
+    skills_dir = os.path.join(temp_dir, "skills")
+    os.makedirs(skills_dir, exist_ok=True)
+
+    async def _run():
+        try:
+            async_gen = _yield_mcp_response(
+                chat_model,
+                system_prompt,
+                message_list,
+                mcp_servers,
+                mcp_output_enable,
+                tool_init_params,
+                source_id,
+                source_type,
+                temp_dir,
+                chat_id,
+                extra_tools,
+            )
+            async for chunk in async_gen:
+                result_queue.put(("data", chunk))
+        except Exception as e:
+            maxkb_logger.error(f"Exception: {e}", exc_info=True)
+            result_queue.put(("error", e))
+        finally:
+            result_queue.put(("done", None))
+
+    # 在全局循环中调度任务
+    asyncio.run_coroutine_threadsafe(_run(), loop)
+
+    while True:
+        msg_type, data = result_queue.get()
+        if msg_type == "done":
+            # 清理临时文件夹
+            shutil.rmtree(temp_dir, ignore_errors=True)
+            break
+        if msg_type == "error":
+            # 清理临时文件夹
+            shutil.rmtree(temp_dir, ignore_errors=True)
+            raise data
+        yield data
+
+
+def build_schema(fields: dict):
+    return create_model("dynamicSchema", **fields)
+
+
+def get_type(_type: str):
+    if _type == "float":
+        return float
+    if _type == "string":
+        return str
+    if _type == "int":
+        return int
+    if _type == "dict":
+        return dict
+    if _type == "array":
+        return list
+    if _type == "boolean":
+        return bool
+    return object
+
+
+def get_workflow_args(tool, qv):
+    for node in qv.work_flow.get("nodes"):
+        if node.get("type") == "tool-base-node":
+            input_field_list = node.get("properties").get("user_input_field_list")
+            return build_schema(
+                {
+                    field.get("field"): (
+                        get_type(field.get("type")),
+                        Field(..., required=True, description=field.get("desc"))
+                        if field.get("is_required")
+                        else Field(default=None, required=False, description=field.get("desc")),
+                    )
+                    for field in input_field_list
+                }
+            )
+
+    return build_schema({})
+
+
+def _save_workflow_tool_record(
+    tool_record_id, tool_id, workspace_id, source_type, source_id, wf_manage, aggregation, parameters, start_time, error
+):
+    """
+    工具工作流执行结束后落库执行记录（替代旧引擎 ToolWorkflowPostHandler.handler）。
+    实实行（非调试）直接插入 ToolRecord，字段与工具记录查询端点保持一致。
+    """
+    workflow = wf_manage.workflow
+    base_node = workflow.get_node("tool-base-node")
+    input_field_list = base_node.properties.get("user_input_field_list", []) if base_node else []
+    output_field_list = base_node.properties.get("user_output_field_list", []) if base_node else []
+    input_data = {f.get("field"): parameters.get(f.get("field")) for f in input_field_list}
+    # 新引擎工具输出统一收口于全局 output 上下文（tool-start-node 初始化、变量赋值节点写入）
+    output = wf_manage.context.get("output", {})
+    details = wf_manage.get_details()
+    if error:
+        state = State.FAILURE
+    else:
+        has_fail = any((d or {}).get("status") == Status.FAIL.value for d in (details or []))
+        state = State.FAILURE if has_fail else State.SUCCESS
+    ToolRecord(
+        id=tool_record_id,
+        tool_id=tool_id,
+        workspace_id=workspace_id,
+        source_type=source_type,
+        source_id=source_id,
+        state=state,
+        run_time=time.time() - start_time,
+        meta={
+            "input_field_list": input_field_list,
+            "output_field_list": output_field_list,
+            "input": input_data,
+            "output": output,
+            "details": details,
+            "answer_text_list": aggregation.get_contents(),
+        },
+    ).save()
+
+
+def get_workflow_func(source_type, source_id, tool, qv, workspace_id):
+    tool_id = tool.id
+
+    def inner(**kwargs):
+        # 使用新工作流引擎执行工具工作流，方式与 tool_workflow_lib_node 保持一致。
+        from application.workflow.common import WorkflowType, new_instance
+        from application.workflow.nodes import get_node_class
+        from application.workflow.workflow_manage import CallBack, WorkflowManage
+
+        tool_record_id = str(uuid.uuid7())
+        sub_workflow = new_instance(qv.work_flow, WorkflowType.TOOL)
+        start_time = time.time()
+        sub_parameters = {
+            "chat_record_id": tool_record_id,
+            "tool_id": str(tool_id),
+            "stream": True,
+            "workspace_id": workspace_id,
+            "default_model_setting": qv.default_model_setting or {},
+            **kwargs,
+        }
+
+        # WorkflowManage.run() 在后台线程异步执行节点，完成时机由 on_complete 回调驱动，
+        # 而 inner 作为 LangChain 同步工具函数必须阻塞到子工作流结束再返回其输出。
+        aggregation = AggregationManager()
+        done_event = threading.Event()
+        result_holder = {"output": {}, "error": None}
+
+        def on_next(wf_manage, content):
+            # 逐块聚合，用于执行记录的 answer_text_list（不直接转发给上游）
+            aggregation.aggregate(content)
+
+        def on_complete(wf_manage, error):
+            try:
+                # 工具工作流输出统一写入 context['output']
+                result_holder["output"] = dict(wf_manage.context.get("output", {}) or {})
+                # 执行结束落库工具执行记录
+                _save_workflow_tool_record(
+                    tool_record_id,
+                    tool_id,
+                    workspace_id,
+                    source_type,
+                    source_id,
+                    wf_manage,
+                    aggregation,
+                    sub_parameters,
+                    start_time,
+                    error,
+                )
+            finally:
+                result_holder["error"] = error
+                done_event.set()
+
+        call_back = CallBack(on_next, on_complete)
+
+        def get_start_node_fn(wf, wm):
+            start_node = wf.get_node("tool-start-node")
+            node_class = get_node_class("tool-start-node", WorkflowType.TOOL)
+            return node_class(start_node, wm, lambda n: n.properties.get("node_data", {}))
+
+        sub_manage = WorkflowManage(
+            workflow=sub_workflow,
+            parameters=sub_parameters,
+            workflow_type=WorkflowType.TOOL,
+            call_back=call_back,
+            get_start_node=get_start_node_fn,
+        )
+        sub_manage.start_node.workflow_manage = sub_manage
+        sub_manage.run()
+        done_event.wait()
+        if result_holder["error"]:
+            raise result_holder["error"]
+        return result_holder["output"]
+
+    return inner
+
+
+def get_workflow_tools(source_type, source_id, tool_workflow_ids, workspace_id):
+    tools = QuerySet(Tool).filter(
+        id__in=tool_workflow_ids, is_active=True, tool_type=ToolType.WORKFLOW, workspace_id=workspace_id
+    )
+    latest_subquery = ToolWorkflowVersion.objects.filter(tool_id=OuterRef("tool_id")).order_by("-create_time")
+
+    qs = ToolWorkflowVersion.objects.filter(
+        tool_id__in=[t.id for t in tools], id=Subquery(latest_subquery.values("id")[:1])
+    )
+    qd = {q.tool_id: q for q in qs}
+    results = []
+    for tool in tools:
+        qv = qd.get(tool.id)
+        func = get_workflow_func(source_type, source_id, tool, qv, workspace_id)
+        args = get_workflow_args(tool, qv)
+        tool = StructuredTool.from_function(
+            func=func,
+            name=tool.name,
+            description=tool.desc,
+            args_schema=args,
+        )
+        results.append(tool)
+
+    return results
diff --git a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py
index 4db9f18b09b..e468eefd5ae 100644
--- a/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py
+++ b/apps/application/workflow/nodes/ai_chat_node/ai_chat_node.py
@@ -19,7 +19,7 @@
 from rest_framework import serializers
 
 from application.workflow.message.aggregator import AggregationManager
-from application.flow.tools import get_tools, mcp_response_generator
+from application.workflow.nodes.ai_chat_node.agent import get_workflow_tools, mcp_response_generator
 from application.models import Application, ApplicationAccessToken, ApplicationApiKey
 from application.workflow.common import WorkflowType
 from application.workflow.i_node import INode
@@ -508,7 +508,7 @@ def _handle_mcp(
             source_id = self.get_workflow_parameters().get("application_id")
             source_type = "APPLICATION"
 
-        tools = get_tools(source_type, chat_id, tool_ids, workspace_id)
+        tools = get_workflow_tools(source_type, chat_id, tool_ids, workspace_id)
         if tool_ids and len(tool_ids) > 0:
             custom_tools_map = {
                 str(t.id): t for t in QuerySet(Tool).filter(id__in=tool_ids, tool_type=ToolType.CUSTOM, is_active=True)