跳到主要内容

🔧 Tool 开发

⚠️ 重要安全警告

工作区中的 Tools 会在您的服务器上执行任意 Python 代码。 请仅安装来自信任源的工具，在导入前仔细审核代码，并将工作区访问权限仅限于可信的管理员。授予用户创建或导入 Tools 的权限，等同于向其提供了服务器的终端 shell 访问权限。有关完整详情，请参阅 Plugin 安全警告。

编写自定义工具箱​

工具箱（Toolkits）定义在单个 Python 文件中，其顶部配有一个用于声明元数据的文档字符串，内部包含一个 Tools 类。

建议使用异步函数以实现未来兼容性

Tool 里的方法通常应当定义为 async（异步），以确保与未来的 Open WebUI 版本兼容。后端正在逐步走向完全的异步执行，同步函数可能会阻塞执行，或在未来的版本中导致各种问题。

顶级文档字符串示例​

"""
title: String Inverse
author: Your Name
author_url: https://website.com
git_url: https://github.com/username/string-reverse.git
description: This tool calculates the inverse of a string
required_open_webui_version: 0.4.0
requirements: langchain-openai, langgraph, ollama, langchain_ollama
version: 0.4.0
licence: MIT
"""

Tools 类​

Tools 必须定义在一个名为 Tools 的类的方法中，内部可以包含可选的 Valves 和 UserValves 子类。示例如下：

class Tools:
    def __init__(self):
        """初始化 Tool。"""
        self.valves = self.Valves()

    class Valves(BaseModel):
        api_key: str = Field("", description="在此处填写您的 API 密钥")

    async def reverse_string(self, string: str) -> str:
        """
        反转输入的字符串。
        :param string: 要反转的字符串
        """
        # valves 配置项的使用示例
        if self.valves.api_key != "42":
            return "API 密钥错误"
        return string[::-1]

类型提示​

每个 tool 的参数必须具有类型提示（type hints）。这些类型也可以是嵌套的，例如 queries_and_docs: list[tuple[str, int]]。这些类型提示会被用来生成发送给模型的 JSON schema 规范。没有任何类型提示的 tools 在模型调用时的一致性会大幅降低。

Valves 和 UserValves（可选，但强烈推荐）​

Valves 和 UserValves 用于指定 Tool 的可自定义配置设置。您可以阅读专门的 Valves & UserValves 页面 获取更多信息。

可选参数​

以下是您的 tools 可以依赖的可选注入参数列表：

  • __event_emitter__：用于发射事件（见下文事件发射器章节）。
  • __event_call__：与事件发射器相同，但可用于收集用户交互反馈。服务器端对事件调用的超时时间可通过 WEBSOCKET_EVENT_CALLER_TIMEOUT 进行配置（默认值：300 秒）。
  • __user__：包含用户信息的字典。它还在 __user__["valves"] 中包含了用户的 UserValves 对象。
  • __metadata__：包含聊天元数据的字典。
  • __messages__：历史消息列表。
  • __files__：上传并附加的文件列表。
  • __model__：包含模型信息的字典。
  • __oauth_token__：包含用户有效的、能自动刷新的 OAuth 令牌 Payload 的字典。这是访问用户令牌以进行身份验证 API 调用的全新、推荐且最安全的方式。该字典通常包含 access_token、id_token 以及其他提供商特定的数据。

如需了解有关 __oauth_token__ 的更多信息以及如何配置该令牌以发送至 tools，请查阅 环境变量配置页面 中的 OAuth 章节以及 SSO 认证文档。

您只需将这些参数声明为您的 Tool 类中任何方法的参数即可，如上面示例中的 __user__。

在 Tool 中使用 OAuth 令牌​

在构建需要代表用户与外部 API 进行交互的 tools 时，您现在可以直接获取他们的 OAuth 令牌。这避免了脆弱且容易出错的 Cookie 爬取，并能确保令牌始终有效。

示例：使用用户的 Access Token 调用外部 API 的 tool。

import httpx
from typing import Optional

class Tools:
    # ... 其他类初始化代码 ...

    async def get_user_profile_from_external_api(self, __oauth_token__: Optional[dict] = None) -> str:
        """
        使用用户的 OAuth 访问令牌从安全的外部 API 获取用户个人资料数据。

        :param __oauth_token__: 由 Open WebUI 注入，包含用户的 token 数据。
        """
        if not __oauth_token__ or "access_token" not in __oauth_token__:
            return "Error: 用户未通过 OAuth 进行身份验证，或 token 不可用。"

        access_token = __oauth_token__["access_token"]

        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json"
        }

        try:
            async with httpx.AsyncClient() as client:
                response = await client.get("https://api.my-service.com/v1/profile", headers=headers)
                response.raise_for_status() # 对非 2xx 状态码抛出异常
                return f"API Response: {response.json()}"
        except httpx.HTTPStatusError as e:
            return f"Error: 无法从 API 获取数据。状态码: {e.response.status_code}"
        except Exception as e:
            return f"发生未知错误: {e}"

事件发射器 (Event Emitters)​

事件发射器用于在聊天界面中动态添加额外的信息。与 Filter 的 outlet 类似，事件发射器能够在聊天中追加内容。不同于 Filter 的 outlet，它们不能剥离现有信息。此外，在 Tool 执行的任何阶段都可以随时激活事件发射器。

请专门针对 Native 模式设计 Tools

Default 模式属于遗留模式，目前已不再受支持 — 详情请查阅 Tool 调用模式指南。请编写您的 tools，使其能在 Native (Agentic) 模式下正确工作，因为这是未来唯一受支持的模式。下方的事件发射器兼容性矩阵仅为了历史参考，以及为了尚未迁移的既有 tools 维护者保留了对 Default 模式行为的记录 — 但新的 tools 不应该再依赖仅在 Default 模式下可用的事件类型。如果您工具的交互设计从根本上需要仅在 Default 模式下运作的事件类型（如 message、chat:message:delta、chat:message、以及流式传输中途的 replace），请围绕 Native 模式兼容的事件（如 status、notification、citation、chat:message:files、confirmation、chat:message:follow_ups）重新设计交互体验，而不是强求用户将他们的模型切换到不推荐的遗留模式。

事件发射器的行为在两种函数调用模式之间存在明显差异。函数调用模式由 function_calling 参数控制：

  • Native 模式 (Agentic 模式) (function_calling = "native") — 唯一受支持的模式。使用模型的结构化工具调用（tool-call）API。事件发射器功能范围受限 — 请参考下方兼容性矩阵，以明确支持哪些事件类型。
  • Default 模式 (function_calling = "default") — 遗留的、基于 Prompt 注入的模式。虽支持全套事件发射器，但模式本身已不再推荐，新部署时不应当选择此模式。

有关完整的模式政策、模型要求和配置方法，请参阅 Tool 调用模式指南。简而言之：请使用 Native 模式；下方矩阵说明了哪些事件类型在此模式下可用。

函数调用模式配置​

您可以在两处配置函数调用模式：

  1. 管理员级别：前往 管理员面板 > 设置 > 模型 > 模型特定设置 > 高级参数 > Function Calling（设为 "Default" 或 "Native"）。
  2. 单次请求：在聊天控制项 > 高级参数中设置 params.function_calling = "native" 或 "default"。

如果模型似乎无法调用该工具，请确保已启用该工具（可以在模型页面上启用，也可以在聊天输入框旁的 + 号中启用）。

Native 模式与内置工具

在编写自定义 tools 时，请注意当启用 Native 模式后，Open WebUI 也会提供内置的系统工具。有关内置工具、函数调用模式和模型要求的详细信息，请参阅 Tool 调用模式指南。

完整的事件类型兼容性矩阵​

以下是各类事件类型在不同函数调用模式下的表现明细：

事件类型Default 模式表现Native 模式表现状态
status✅ 完全支持 - 在 tool 执行期间更新状态历史✅ 表现一致 - 追踪函数执行状态兼容 (COMPATIBLE)
message✅ 完全支持 - 在流式传输中追加增量内容❌ 失效 - 会被 native 的完整生成快照覆盖不兼容 (INCOMPATIBLE)
chat:completion✅ 完全支持 - 处理流式响应和生成数据⚠️ 受限 - 承载函数结果但可能覆盖 tool 里的更新部分兼容
chat:message:delta✅ 完全支持 - 在执行期间流式输出 delta 增量内容❌ 失效 - 内容会被 native 的完整生成快照覆盖不兼容 (INCOMPATIBLE)
chat:message✅ 完全支持 - 干净利落地替换整条消息内容❌ 失效 - 会被后续的 native 生成快照覆盖不兼容 (INCOMPATIBLE)
replace✅ 完全支持 - 精确控制并替换内容❌ 失效 - 替换后的内容会立刻被后续生成快照覆盖不兼容 (INCOMPATIBLE)
chat:message:files / files✅ 完全支持 - 处理消息中的文件附件✅ 表现一致 - 处理来自函数输出的文件兼容 (COMPATIBLE)
chat:message:error✅ 完全支持 - 展示错误通知✅ 表现一致 - 展示函数调用错误兼容 (COMPATIBLE)
chat:message:follow_ups✅ 完全支持 - 显示后续跟进问题建议✅ 表现一致 - 展示函数生成的后续跟进问题兼容 (COMPATIBLE)
chat:title✅ 完全支持 - 动态更新聊天标题✅ 表现一致 - 基于函数交互更新标题兼容 (COMPATIBLE)
chat:tags✅ 完全支持 - 修改聊天标签✅ 表现一致 - 通过函数输出管理标签兼容 (COMPATIBLE)
chat:tasks:cancel✅ 完全支持 - 取消正在进行的任务✅ 表现一致 - 取消 native 函数执行兼容 (COMPATIBLE)
citation / source✅ 完全支持 - 处理包含完整元数据的引文✅ 表现一致 - 处理函数生成的引文兼容 (COMPATIBLE)
notification✅ 完全支持 - 显示气泡提示通知 (Toast)✅ 表现一致 - 展示函数执行通知兼容 (COMPATIBLE)
confirmation✅ 完全支持 - 请求用户确认操作✅ 表现一致 - 确认函数执行兼容 (COMPATIBLE)
execute✅ 完全支持 - 动态执行代码✅ 表现一致 - 运行函数生成的代码兼容 (COMPATIBLE)
input✅ 完全支持 - 呈现完整的 UI 请求用户输入✅ 表现一致 - 为函数收集输入兼容 (COMPATIBLE)

为什么 Native 模式会破坏某些事件类型​

在 Native 模式下，服务器会根据流式输出的模型内容构建内容块，并反复发射带有完整序列化内容快照的 "chat:completion" 事件。前端客户端会将这些快照视为权威内容，并完全替换消息内容，这实际上会直接抹除工具之前发射的任何更新事件（如 message、chat:message 或 replace）。

技术细节：

  • middleware.py 将 tools 直接添加到 form data 中以供 native 模型处理。
  • 流式传输处理器通过 chat:completion 事件反复发射完整的内容快照。
  • 客户端的 chatCompletionEventHandler 会将快照作为整块内容进行彻底替换：message.content = content。
  • 这会导致 tool 发射的内容更新出现闪烁并随即消失。

最佳实践与建议​

对于需要实时 UI 更新的 Tools（仅在 Default 模式下生效）：

class Tools:
    def __init__(self):
        # 提示用户关于函数调用模式的要求
        self.description = "该工具需要 Default 函数调用模式以获取完整的功能体验"

    async def interactive_tool(self, prompt: str, __event_emitter__=None) -> str:
        """
        ⚠️ 该工具需要 function_calling = "default" 才能正确发射事件
        """
        if not __event_emitter__:
            return "事件发射器不可用 - 请确保已启用 Default 函数调用模式"

        # 在 Default 模式下可以安全地使用 message 事件
        await __event_emitter__({
            "type": "message",
            "data": {"content": "正在处理步骤 1..."}
        })
        # ... 工具的其余逻辑

对于必须同时兼容两种模式的 Tools：

async def universal_tool(self, prompt: str, __event_emitter__=None, __metadata__=None) -> str:
    """
    设计为同时兼容 Default 和 Native 函数调用模式的工具
    """
    # 检测我们是否处于 native 模式（这是一种粗略的启发式判断）
    is_native_mode = __metadata__ and __metadata__.get("params", {}).get("function_calling") == "native"

    if __event_emitter__:
        if is_native_mode:
            # 在 native 模式下，仅使用兼容的事件类型
            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在 Native 模式下处理...", "done": False}
            })
        else:
            # 在 default 模式下，可以使用全部事件功能
            await __event_emitter__({
                "type": "message",
                "data": {"content": "正在处理中，完全的事件支持已启用..."}
            })

    # ... 此处为具体的 tool 逻辑

    if __event_emitter__:
        await __event_emitter__({
            "type": "status",
            "data": {"description": "处理成功完成", "done": True}
        })

    return "工具执行完毕"

解决事件发射器问题​

出现 Native 模式冲突的症状：

  • 工具发送的 message 内容短暂显示后立刻消失。
  • 工具执行期间内容发生闪烁。
  • message 或 replace 事件似乎被完全忽略。
  • status 更新正常工作，但内容更新无法持久保留。

解决方案：

  1. 切换到 Default 模式：在模型设置中将 function_calling 从 "native" 变更为 "default"。
  2. 使用兼容的事件类型：在 native 模式下，请坚持使用 status、citation、notification 以及其他兼容的事件类型。
  3. 实现模式检测：增加检测函数调用模式的逻辑，并相应地调整事件的使用。
  4. 考虑混合兼容方案：对核心功能使用兼容的事件，并在其他地方优雅降级。

调试您的事件发射器：

async def debug_events_tool(self, __event_emitter__=None, __metadata__=None) -> str:
    """调试工具，用以测试事件发射器的功能"""

    if not __event_emitter__:
        return "没有可用的事件发射器"

    # 测试各种事件类型
    test_events = [
        {"type": "status", "data": {"description": "正在测试 status 事件", "done": False}},
        {"type": "message", "data": {"content": "正在测试 message 事件（在 native 模式下可能会失效）"}},
        {"type": "notification", "data": {"content": "正在测试 notification 通知事件"}},
    ]

    mode_info = "Unknown"
    if __metadata__:
        mode_info = __metadata__.get("params", {}).get("function_calling", "default")

    await __event_emitter__({
        "type": "status",
        "data": {"description": f"函数调用模式: {mode_info}", "done": False}
    })

    for i, event in enumerate(test_events):
        await asyncio.sleep(1)  # 错开事件发射的时间
        await __event_emitter__(event)
        await __event_emitter__({
            "type": "status",
            "data": {"description": f"已发送事件 {i+1}/{len(test_events)}", "done": False}
        })

    await __event_emitter__({
        "type": "status",
        "data": {"description": "事件测试完毕", "done": True}
    })

    return f"已在 {mode_info} 模式下完成事件测试。请检查是否有缺失或闪烁的内容。"

具体各种事件类型的具体行为和声明格式如下：

Status 事件 ✅ 完全兼容​

Status 事件在 Default 和 Native 函数调用模式下的表现完全一致。 这是在 tool 执行期间提供实时反馈的最可靠的事件类型。

Status（状态）事件会在消息执行时，向该消息动态地添加实时状态更新。这些更新可以在 tool 执行的任何阶段发出。状态消息会直接呈现在聊天消息内容的上方，对于那些响应时间较长或需要处理大量数据的 tools 而言，这是必不可少的。

基础 Status 事件结构：

await __event_emitter__({
    "type": "status",
    "data": {
        "description": "显示在聊天界面中的提示消息",
        "done": False,        # False = 仍在处理中, True = 已完成
        "hidden": False       # False = 可见, True = 在 done: True 时自动隐藏
    }
})

Status 事件参数：

  • description：展示给用户的状态提示文本。
  • done：布尔值，用于指示该状态是否代表处理已结束。
  • hidden：布尔值，设置为 True 时，一旦状态变更为 done: True，该状态便会自动在界面中隐藏。
基础 Status 示例
async def data_processing_tool(
        self, data_file: str, __user__: dict, __event_emitter__=None
    ) -> str:
        """
        处理大型数据文件，并实时发出状态更新
        ✅ 同时兼容 Default 和 Native 函数调用模式
        """

        if not __event_emitter__:
            return "处理完成（无实时状态更新）"

        # 步骤 1：加载数据
        await __event_emitter__({
            "type": "status",
            "data": {"description": "正在加载数据文件...", "done": False}
        })

        # 模拟加载数据所需时间
        await asyncio.sleep(2)

        # 步骤 2：处理数据
        await __event_emitter__({
            "type": "status",
            "data": {"description": "正在分析 10,000 条记录...", "done": False}
        })

        # 模拟计算所需时间
        await asyncio.sleep(3)

        # 步骤 3：完成
        await __event_emitter__({
            "type": "status",
            "data": {"description": "分析完成！", "done": True, "hidden": False}
        })

        return "数据分析圆满结束。共发现 23 处异常数据。"
带错误处理的高级 Status 示例
async def api_integration_tool(
        self, endpoint: str, __event_emitter__=None
    ) -> str:
        """
        进行外部 API 集成，并执行完善的状态追踪
        ✅ 完美兼容两种函数调用模式
        """

        if not __event_emitter__:
            return "API 集成处理完成（状态不可用）"

        try:
            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在连接到 API...", "done": False}
            })

            # 模拟连接 API
            await asyncio.sleep(1.5)

            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在进行身份验证...", "done": False}
            })

            # 模拟身份验证
            await asyncio.sleep(1)

            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在抓取数据...", "done": False}
            })

            # 模拟数据抓取
            await asyncio.sleep(2)

            # 成功时的状态更新
            await __event_emitter__({
                "type": "status",
                "data": {"description": "API 集成成功！", "done": True}
            })

            return "已成功从 API 检索到 150 条记录"

        except Exception as e:
            # 异常时的状态更新 - 保持可见以供调试
            await __event_emitter__({
                "type": "status",
                "data": {"description": f"发生错误: {str(e)}", "done": True, "hidden": False}
            })

            return f"API 集成失败: {str(e)}"
多步骤进度 Status 示例
async def batch_processor_tool(
        self, items: list, __event_emitter__=None
    ) -> str:
        """
        分批次处理条目，并进行详细的进度追踪
        ✅ 在两种函数调用模式下均能完美运转
        """

        if not __event_emitter__ or not items:
            return "分批处理完成"

        total_items = len(items)
        batch_size = 10
        completed = 0

        for i in range(0, total_items, batch_size):
            batch = items[i:i + batch_size]
            batch_num = (i // batch_size) + 1
            total_batches = (total_items + batch_size - 1) // batch_size

            # 更新当前批次的状态
            await __event_emitter__({
                "type": "status",
                "data": {
                    "description": f"正在处理批次 {batch_num}/{total_batches} ({len(batch)} 项)...",
                    "done": False
                }
            })

            # 模拟批处理计算时间
            await asyncio.sleep(1)

            completed += len(batch)

            # 进度百分比更新
            progress_pct = int((completed / total_items) * 100)
            await __event_emitter__({
                "type": "status",
                "data": {
                    "description": f"当前进度: 已完成 {completed}/{total_items} 项 ({progress_pct}%)",
                    "done": False
                }
            })

        # 最终完成状态
        await __event_emitter__({
            "type": "status",
            "data": {
                "description": f"分批处理圆满结束！共处理 {total_items} 项数据",
                "done": True
            }
        })

        return f"已成功在 {total_batches} 个批次中处理了 {total_items} 项数据"

Message 事件 ⚠️ 仅限 Default 模式​

重要警告

🚨 极其重要：Message 事件与 Native 函数调用模式存在根本性的不兼容！

Message 事件（包括 message、chat:message、chat:message:delta、replace）允许您在 tool 执行的任意阶段追加或修改消息内容。这可以让您轻松嵌入图像、渲染网页、流式展示内容更新并创造丰富的交互式体验。

然而，这些事件类型有着非常严重的兼容性缺陷：

  • ✅ Default 模式：完全支持 — 内容能够干净利落地持久保留并完整显示。
  • ❌ Native 模式：失效 — 内容会被模型最终生成的快照完全覆盖并消失。

为什么 Message 事件在 Native 模式下会失效： Native 函数调用在流式传输时会不断发射 chat:completion 事件并携带完整的内容快照，这会覆盖并彻底替换之前的消息内容，导致由工具发射的任何消息内容更新产生闪烁并最终化为乌有。

安全的 Message 事件结构（仅限 Default 模式）：

await __event_emitter__({
    "type": "message",  # 也支持: "chat:message", "chat:message:delta", "replace"
    "data": {"content": "这部分内容将被追加或替换到聊天消息中"},
    # 注意：message 类型的事件不需要声明 "done" 状态
})

Message 事件细分类别：

  • message / chat:message:delta：将新内容追加到既有消息的尾部。
  • chat:message / replace：将整条消息内容彻底替换。
  • 两种类型在 Native 模式下均会被直接覆盖抹去。
安全的内容流式追加示例（仅限 Default 模式）
async def streaming_content_tool(
        self, query: str, __event_emitter__=None, __metadata__=None
    ) -> str:
        """
        在处理期间流式传输内容更新
        ⚠️ 需要声明 function_calling = "default" - 在 Native 模式下不可用！
        """

        # 粗略检测当前的函数调用模式
        mode = "default"
        if __metadata__:
            mode = __metadata__.get("params", {}).get("function_calling", "default")

        if mode == "native":
            return "❌ 本工具需要 Default 函数调用模式。由于 Native 模式的内容覆写缺陷，这里不支持进行内容流式追加。"

        if not __event_emitter__:
            return "事件发射器不可用"

        # 逐步流式追加的文本块
        content_chunks = [
            "🔍 **第一阶段：收集研究**\n正在为您搜集该问题的背景信息...\n\n",
            "📊 **第二阶段：分析整理**\n正在分析搜集到的数据特征与规律...\n\n",
            "✨ **第三阶段：提炼洞察**\n正在提炼核心结论与优化建议...\n\n",
            "📝 **第四阶段：整理报告**\n正在编译并排版最终的分析报告...\n\n"
        ]

        accumulated_content = ""

        for i, chunk in enumerate(content_chunks):
            accumulated_content += chunk

            # 将本块内容追加至当前消息中
            await __event_emitter__({
                "type": "message",
                "data": {"content": chunk}
            })

            # 更新当前执行进度状态
            await __event_emitter__({
                "type": "status",
                "data": {
                    "description": f"正在处理第 {i+1}/{len(content_chunks)} 阶段...",
                    "done": False
                }
            })

            # 模拟各个阶段的工作耗时
            await asyncio.sleep(2)

        # 最终完成
        await __event_emitter__({
            "type": "status",
            "data": {"description": "报告流式输出完毕！", "done": True}
        })

        return "所有研究阶段均已顺利完成。"
动态内容整体替换示例（仅限 Default 模式）
async def live_dashboard_tool(
        self, __event_emitter__=None, __metadata__=None
    ) -> str:
        """
        通过整体替换内容，创建一个实时更新的看板
        ⚠️ 仅能在 Default 函数调用模式下工作
        """

        # 确认当前不是 Native 模式
        mode = __metadata__.get("params", {}).get("function_calling", "default") if __metadata__ else "default"

        if mode == "native":
            return """
❌ **Native 模式不兼容警告**

本看板工具无法在 Native 模式下正常工作，因为：
- 内容整体替换事件会被模型最终生成的快照完全覆写
- 实时更新的内容会出现闪烁并最终消失
- 界面中无法持久保存实时的监控数据

**解决方案**：请前往模型设置 → 高级参数 → 将“函数调用”切换为“Default”。
"""

        if not __event_emitter__:
            return "看板创建成功（静态模式 - 无实时数据更新）"

        # 初始看板样式
        initial_dashboard = """
# 📊 系统实时状态看板

## 运行状态: 🟡 正在初始化...

### 实时指标:
- **CPU 使用率**: 加载中...
- **内存占用**: 加载中...
- **活跃用户数**: 加载中...
- **系统响应时间**: 加载中...

---
*上次更新时间: 正在初始化...*
"""

        await __event_emitter__({
            "type": "replace",
            "data": {"content": initial_dashboard}
        })

        # 模拟实时数据流变化
        updates = [
            {
                "status": "🟢 正常运行",
                "cpu": "23%",
                "memory": "64%",
                "users": "1,247",
                "response": "145ms"
            },
            {
                "status": "🟢 优异运行",
                "cpu": "18%",
                "memory": "61%",
                "users": "1,352",
                "response": "132ms"
            },
            {
                "status": "🟡 负载偏高",
                "cpu": "67%",
                "memory": "78%",
                "users": "1,891",
                "response": "234ms"
            }
        ]

        for i, data in enumerate(updates):
            await asyncio.sleep(3)  # 模拟数据采集延迟

            updated_dashboard = f"""
# 📊 系统实时状态看板

## 运行状态: {data['status']}

### 实时指标:
- **CPU 使用率**: {data['cpu']}
- **内存占用**: {data['memory']}
- **活跃用户数**: {data['users']}
- **系统响应时间**: {data['response']}

---
*上次更新时间: {datetime.now().strftime('%H:%M:%S')}*
*更新频次: {i+1}/{len(updates)}*
"""

            # 整体替换消息内容
            await __event_emitter__({
                "type": "replace",
                "data": {"content": updated_dashboard}
            })

            # 更新状态栏提示
            await __event_emitter__({
                "type": "status",
                "data": {"description": f"看板数据已更新 ({i+1}/{len(updates)})", "done": False}
            })

        await __event_emitter__({
            "type": "status",
            "data": {"description": "系统实时监控会话结束", "done": True}
        })

        return "看板监控阶段顺利结束。"
自适应模式的消息处理示例
async def adaptive_content_tool(
        self, content_type: str, __event_emitter__=None, __metadata__=None
    ) -> str:
        """
        基于检测到的函数调用模式，动态自适应工具的执行行为
        ✅ 在两种调用模式下均能保证最佳的体验
        """

        # 检测当前的函数调用模式
        mode = "default"
        if __metadata__:
            mode = __metadata__.get("params", {}).get("function_calling", "default")

        if not __event_emitter__:
            return f"已成功生成 {content_type} 内容（无实时事件发射机制）"

        # 根据不同模式，执行针对性的逻辑
        if mode == "native":
            # 在 Native 模式下，仅使用完全兼容的事件
            await __event_emitter__({
                "type": "status",
                "data": {"description": f"正在 Native 模式下生成 {content_type} 内容...", "done": False}
            })

            await asyncio.sleep(2)

            await __event_emitter__({
                "type": "status",
                "data": {"description": "内容生成完毕", "done": True}
            })

            # 将所有的核心内容作为函数的 return 返回值一次性交付 — 绝不通过 message 事件去流式渲染
            return f"""
# {content_type.title()} 内容展示

**当前运行模式**: Native 函数调用（事件支持受限）

此处为成功生成的内容... 在 Native 模式下，内容应当作为 tool 调用的最终返回值完整呈递给模型，而不可使用 message 事件在中途发射。

*说明：由于 Native 模式的底层限制，本模式下不支持流式追加内容。*
"""

        else:  # Default 模式
            # 可以充分调用 message 事件进行美妙的流式输出
            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在流式生成内容中，全套事件已启用...", "done": False}
            })

            # 逐步流式发射内容
            progressive_content = [
                f"# {content_type.title()} 内容展示\n\n**当前运行模式**: Default 函数调用 ✅\n\n",
                "## 第一节：前言\n正在为您实时流式推送前言部分...\n\n",
                "## 第二节：具体详情\n正在为您实时流式推送详情分析...\n\n",
                "## 第三节：总结\n正在为您流式推送最后的总结报告...\n\n",
                "*✅ 内容全部流式生成完毕！*"
            ]

            for i, chunk in enumerate(progressive_content):
                await __event_emitter__({
                    "type": "message",
                    "data": {"content": chunk}
                })

                await __event_emitter__({
                    "type": "status",
                    "data": {"description": f"正在推送第 {i+1}/{len(progressive_content)} 部分...", "done": False}
                })

                await asyncio.sleep(1.5)

            await __event_emitter__({
                "type": "status",
                "data": {"description": "流式生成全部结束！", "done": True}
            })

            return "内容已全部在上方以流式追加的机制安全呈递。"

Citations 引文事件 ✅ 完全兼容​

Citations 事件在 Default 和 Native 函数调用模式下的表现完全一致。 该事件类型用于在聊天页面中附加文献参考或数据来源链接，允许用户点击并弹窗查阅原始资料。

如果您的 tool 会从外部资料库、文档检索（RAG）、网络搜索或者关系型数据库中搜集信息，那么 Citation 事件是绝对不可或缺的，它为用户提供了极其优良的数据可追溯性。

Citation 事件结构：

await __event_emitter__({
    "type": "citation",
    "data": {
        "document": [content],                    # 内容文本数组
        "metadata": [                             # 元数据字典数组
            {
                "date_accessed": datetime.now().isoformat(),
                "source": title,
                "author": "作者姓名",             # 可选
                "publication_date": "2024-01-01", # 可选
                "url": "https://source-url.com"   # 可选
            }
        ],
        "source": {"name": title, "url": url}    # 主要的来源链接与名字
    }
})

配置自定义引文的重要步骤： 当您打算自己在工具中手动控制并向前端发射引文事件时，您必须在您的 Tools 类中将自动引文检测关闭：

def __init__(self):
    self.citation = False  # ⚠️ 必须设置 — 防止系统的自动引文覆盖您自己定义的引文
注意

⚠️ 关键引文警告： 如果您将 self.citation = True（或者没有手动初始化为 False），Open WebUI 内部的自动引文逻辑将会完全覆盖并抹除您发送的任何自定义引文事件。因此，当需要发送自定义引文时，请时刻确保将此变量设为 False。

基础 Citation 示例
class Tools:
    def __init__(self):
        self.citation = False  # 关闭系统的自动引文检测

    async def research_tool(
            self, topic: str, __event_emitter__=None
        ) -> str:
        """
        搜集特定主题的背景信息，并提供规范的引文
        ✅ 在 Default 和 Native 模式下的表现完全一致
        """

        if not __event_emitter__:
            return "研究结束（引文机制不可用）"

        # 模拟搜集到的文献资料
        sources = [
            {
                "title": "先进 AI 系统前沿研究",
                "url": "https://example.com/ai-systems",
                "content": "近年来，人工智能系统架构发生了长足而深刻的演变...",
                "author": "Dr. Jane Smith",
                "date": "2024-03-15"
            },
            {
                "title": "机器学习算法基石",
                "url": "https://example.com/ml-fundamentals",
                "content": "机器学习的核心驱动力与基本原则涵盖了...",
                "author": "Prof. John Doe",
                "date": "2024-02-20"
            }
        ]

        # 针对每个数据源发射引文事件
        for source in sources:
            await __event_emitter__({
                "type": "citation",
                "data": {
                    "document": [source["content"]],
                    "metadata": [
                        {
                            "date_accessed": datetime.now().isoformat(),
                            "source": source["title"],
                            "author": source["author"],
                            "publication_date": source["date"],
                            "url": source["url"]
                        }
                    ],
                    "source": {
                        "name": source["title"],
                        "url": source["url"]
                    }
                }
            })

        return f"针对主题 '{topic}' 的研究搜集工作已结束。共为您检索到 {len(sources)} 篇权威学术文献，引文链接已附加至页面。"
高级多源文献引文示例
async def comprehensive_analysis_tool(
        self, query: str, __event_emitter__=None
    ) -> str:
        """
        进行包含各种复杂类型的多源数据分析与引文呈递
        ✅ 在所有的函数调用模式下均能完美兼容
        """

        if not __event_emitter__:
            return "分析结束"

        # 包含学术、网络、行业报告等不同种类的多元文献
        research_sources = {
            "academic": [
                {
                    "title": "现代人工智能中的神经网络架构演变",
                    "authors": ["Dr. Sarah Chen", "Prof. Michael Rodriguez"],
                    "journal": "Journal of AI Research",
                    "volume": "Vol. 45, Issue 2",
                    "pages": "123-145",
                    "doi": "10.1000/182",
                    "date": "2024-01-15",
                    "content": "本篇文献系统性地剖析了深度神经网络拓扑结构的发展历程..."
                }
            ],
            "web_sources": [
                {
                    "title": "2024 年行业 AI 落地与实施趋势白皮书",
                    "url": "https://tech-insights.com/ai-trends-2024",
                    "site_name": "TechInsights",
                    "published": "2024-03-01",
                    "content": "最新的行业问卷调查显示，目前约有 78% 的受访企业正在将 AI 技术整合进核心业务流程..."
                }
            ],
            "reports": [
                {
                    "title": "全球人工智能市场 2024 年度展望研究报告",
                    "organization": "International Tech Research Institute",
                    "report_number": "ITRI-2024-AI-001",
                    "date": "2024-02-28",
                    "content": "预计到 2030 年，全球人工智能产业的整体产值将达到 1.8 万亿美元..."
                }
            ]
        }

        citation_count = 0

        # 处理学术期刊引文
        for source in research_sources["academic"]:
            citation_count += 1
            await __event_emitter__({
                "type": "citation",
                "data": {
                    "document": [source["content"]],
                    "metadata": [
                        {
                            "date_accessed": datetime.now().isoformat(),
                            "source": source["title"],
                            "authors": source["authors"],
                            "journal": source["journal"],
                            "volume": source["volume"],
                            "pages": source["pages"],
                            "doi": source["doi"],
                            "publication_date": source["date"],
                            "type": "academic_journal"
                        }
                    ],
                    "source": {
                        "name": f"{source['title']} - {source['journal']}",
                        "url": f"https://doi.org/{source['doi']}"
                    }
                }
            })

        # 处理网页资讯引文
        for source in research_sources["web_sources"]:
            citation_count += 1
            await __event_emitter__({
                "type": "citation",
                "data": {
                    "document": [source["content"]],
                    "metadata": [
                        {
                            "date_accessed": datetime.now().isoformat(),
                            "source": source["title"],
                            "site_name": source["site_name"],
                            "publication_date": source["published"],
                            "url": source["url"],
                            "type": "web_article"
                        }
                    ],
                    "source": {
                        "name": source["title"],
                        "url": source["url"]
                    }
                }
            })

        # 处理分析报告引文
        for source in research_sources["reports"]:
            citation_count += 1
            await __event_emitter__({
                "type": "citation",
                "data": {
                    "document": [source["content"]],
                    "metadata": [
                        {
                            "date_accessed": datetime.now().isoformat(),
                            "source": source["title"],
                            "organization": source["organization"],
                            "report_number": source["report_number"],
                            "publication_date": source["date"],
                            "type": "research_report"
                        }
                    ],
                    "source": {
                        "name": f"{source['title']} - {source['organization']}",
                        "url": f"https://reports.example.com/{source['report_number']}"
                    }
                }
            })

        return f"""
# 复杂数据分析已完成

针对您提出的问题 '{query}' 的深度跨平台分析已成功结束。本报告共交叉索引并引用了 {citation_count} 篇极具说服力的核心权威资料：

- **{len(research_sources['academic'])} 篇** 学术期刊出版物
- **{len(research_sources['web_sources'])} 篇** 行业专业网络评测白皮书
- **{len(research_sources['reports'])} 篇** 全球顶级咨询机构行研报告

所有的引文参考及链接已经安全挂载至页面上方，您可以随时点击链接以在弹窗中查阅原始文献细节。
"""
数据库查询结果引文示例
async def database_query_tool(
        self, sql_query: str, __event_emitter__=None
    ) -> str:
        """
        检索内部数据库，并提供详尽的数据审计与数据溯源引文
        ✅ 在所有的函数调用模式下均能完美运转
        """

        if not __event_emitter__:
            return "数据库查询执行完毕"

        # 模拟检索出的关系型数据库数据及其元数据
        query_results = [
            {
                "record_id": "USR_001247",
                "data": "John Smith, 资深软件工程师, 2023-01-15 入职",
                "table": "employees",
                "last_updated": "2024-03-10T14:30:00Z",
                "updated_by": "admin_user"
            },
            {
                "record_id": "USR_001248",
                "data": "Jane Wilson, 核心产品经理, 2023-02-20 入职",
                "table": "employees",
                "last_updated": "2024-03-08T09:15:00Z",
                "updated_by": "hr_system"
            }
        ]

        # 针对每一条核心数据行，在页面中生成数据引文
        for i, record in enumerate(query_results):
            await __event_emitter__({
                "type": "citation",
                "data": {
                    "document": [f"数据库真实记录: {record['data']}"],
                    "metadata": [
                        {
                            "date_accessed": datetime.now().isoformat(),
                            "source": f"内部数据表: {record['table']}",
                            "record_id": record['record_id'],
                            "last_updated": record['last_updated'],
                            "updated_by": record['updated_by'],
                            "query": sql_query,
                            "type": "database_record"
                        }
                    ],
                    "source": {
                        "name": f"员工记录 {record['record_id']} - {record['table']}",
                        "url": f"database://internal/tables/{record['table']}/{record['record_id']}"
                    }
                }
            })

        return f"""
# 数据库查询数据汇总

本次执行的 SQL 检索语句: `{sql_query}`

已成功为您从库中查询并抓取到 **{len(query_results)}** 条权威的数据记录，每条记录均已附带完整的数据审计引文，审计指标包括：
- 关系表来源与数据唯一标识符（Record ID）
- 数据的最新修改戳记
- 修改数据的归属系统或操作账号
- 完整的系统审计追踪机制

所有的查询来源均已建立规范的内链数据引文，以供随时在界面上查看细节以确保数据真实可信。
"""

其他完全兼容的事件类型 ✅​

以下事件类型在 Default 模式和 Native 模式下的运行效果完全一致，您可以放心调用：

气泡通知事件 (Notification Events)

await __event_emitter__({
    "type": "notification",
    "data": {"content": "成功发射 Toast 气泡通知！"}
})

文件挂载事件 (File Events)

await __event_emitter__({
    "type": "files", # 亦可写为 "chat:message:files"
    "data": {"files": [{"name": "analysis_report.pdf", "url": "/files/analysis_report.pdf"}]}
})

后续问题推荐事件 (Follow-up Events)

await __event_emitter__({
    "type": "chat:message:follow_ups",
    "data": {"follow_ups": ["我想详细剖析指标 X", "请告诉我更多有关 Y 的细节"]}
})

更新会话标题事件 (Title Update Events)

await __event_emitter__({
    "type": "chat:title",
    "data": {"title": "新的会话主题"}
})

打标签事件 (Tag Events)

await __event_emitter__({
    "type": "chat:tags",
    "data": {"tags": ["数据分析", "学术研究", "处理完毕"]}
})

报错提示事件 (Error Events)

await __event_emitter__({
    "type": "chat:message:error",
    "data": {"content": "捕获到未预期的系统异常提示"}
})

请求确认事件 (Confirmation Events)

await __event_emitter__({
    "type": "confirmation",
    "data": {"message": "您确定要继续执行这项高风险的指令吗？"}
})

请求输入事件 (Input Request Events)

await __event_emitter__({
    "type": "input",
    "data": {"prompt": "请补充您的其他详细参数以继续运行："}
})

远程代码执行事件 (Code Execution Events)

await __event_emitter__({
    "type": "execute",
    "data": {"code": "print('成功运行由工具动态生成的代码段！')"}
})

详尽的函数调用模式指南​

请针对 Native 模式设计插件 — 遗留的 Default 模式已不再受支持

Default 模式由于底层限制，已经属于遗留技术，目前在生产环境中不再获得官方支持。所有新编写的 tools 必须全盘契合 Native 模式的设计。本节所列的细节对比仅作为老工具迁移时的底层技术对齐，而绝对不是推荐您在新项目开发中选择已落后的 Default 模式。

如果您的交互体验设计强力绑定了只有 Default 模式才支持的特定事件类型（如中途流式追加内容的 message 事件），请立刻改写您的交互方案，将其围绕 Native 模式原生支持的事件（如 status、notification、citation、chat:message:files、confirmation、chat:message:follow_ups，并且最终由函数方法的 return 来一次性交付完整的内容快照）进行重构。在今后的插件开发中，强行要求用户将模型运行参数切换到落后的遗留模式以运行您的工具是不可被接受的。

函数调用模式技术指标对比 (仅用于迁移时对照参考 — Native 才是唯一的目标运行模式):

技术指标维度Default 模式 (遗留 / 不再支持)Native 模式 (现代 / 唯一受支持)
项目态度❌ 遗留模式，不再维护✅ 新工具开发的强制规范
整体耗时 (Latency)偏高 — 历经多层复杂的 Prompt 注入管道极低 — 基于模型本身的 Tool-call 结构化 API 直连
KV 缓存状态❌ 每轮对话交互均会遭到破坏✅ 完好保存以加速生成
事件发射支持完整事件全量支持仅限特定事件支持（见上表兼容性矩阵）
中途流式追加允许在流传输中途利用事件改写内容会被模型最终完成的完整快照直接覆写并抹除
状态栏更新 / Toast / 引文✅ 支持✅ 完美支持
消息体级别事件 (message, chat:message, replace)✅ 支持❌ 无法持久，会被快照冲刷抹除
系统内置功能性工具 (如 Memory, Notes, Knowledge, Web Search 等)❌ 无法共存✅ 可以完美协同调用

如何围绕 Native 模式进行工具的架构设计：

  • 务必将您工具最终计算出的核心文本，直接以工具方法的返回值 (return value) 形式进行交付。由 return 交付的内容会在模型的 response 中被直接发射和持久渲染，完全不会受到事件被快照覆盖的问题影响。
  • 多多使用 status 和 notification 事件来向用户展示任务进度以及即时的消息提示 — 这些在 Native 模式中均被完全支持。
  • 使用 citation / source 事件来呈献您的数据检索文献引文 — 在 Native 模式下完全兼容。
  • 使用 confirmation / input 并结合 __event_call__ 参数来实现和用户的高级交互确认流程 — 在 Native 模式下完全兼容。
  • 如果需要向用户呈现或发送文件，调用 chat:message:files 事件 — 在 Native 模式下完全兼容。
  • 绝不应该在运行中途去依赖 message、chat:message:delta、chat:message 或 replace 这四种在 Native 模式下会被冲刷失效的事件。如果您的工具包含耗时较长的渐进式进展，请连续发射一连串 status 事件以在状态栏反馈，并将最终汇聚的全部结果作为函数末尾的返回值一次性 return。
将一个原先运行在 Default 模式下的工具迁移到 Native 模式

把 Default 模式的老工具升级为 Native 模式时的通用重构技巧：

  • 抛弃所有在中途使用 message 事件一点点发射或追加内容的代码，将其重写为在工具的最后 return 整体汇聚的完整字符串。
  • 将使用 replace 事件制作的炫酷进度条，更改为在不同阶段持续发射 status 事件（如："description": "第 3 阶段/共 5 阶段 — 正在拉取数据..."）。
  • 将原来使用 chat:message 事件在中途渲染的实时看板，重构为直接在工具最后的 return 返回值中返回一段排版精美的 Markdown 或是 HTML 看板代码。
  • 保持原有的 citation、notification、confirmation 以及文件附件逻辑不变 — 它们在两种模式下的运行效果本就完全一致，无需任何改动。

通用架构自适应重构模板：

async def mode_adaptive_tool(
        self, query: str, __event_emitter__=None, __metadata__=None
    ) -> str:
        """
        能够自动检测当前的运行环境并自适应自调整的通用工具模板
        ✅ 在新老环境里均能保证顺畅的用户体验
        """

        # 检测当前处于哪一种函数调用模式下
        mode = "default"
        if __metadata__:
            mode = __metadata__.get("params", {}).get("function_calling", "default")

        is_native_mode = (mode == "native")

        if not __event_emitter__:
            return "工具执行完毕（缺少事件通知机制）"

        # status 事件在所有环境下都是安全的，放心调用
        await __event_emitter__({
            "type": "status",
            "data": {"description": f"已自动适配到 {mode} 环境下运行...", "done": False}
        })

        # 执行自适应环境分支
        if is_native_mode:
            # Native 模式下：仅调用完全兼容的 status 等事件
            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在使用 Native 原生高能链路计算中...", "done": False}
            })

            # 模拟执行运算
            await asyncio.sleep(1)

            # 核心结果直接交付 return，绝不使用 message 事件流式追加
            result = f"已在 Native 原生极速模式下成功处理完毕您的查询: '{query}'。"

        else:
            # 遗留的 Default 模式下：可以豪迈地调用全套 message 事件进行流式呈递
            await __event_emitter__({
                "type": "message",
                "data": {"content": f"🔍 **开始处理查询**: {query}\n\n"}
            })

            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在流式推送数据分析...", "done": False}
            })

            await asyncio.sleep(1)

            await __event_emitter__({
                "type": "message",
                "data": {"content": "📊 **数据简报**: 所有维度的交叉分析已顺利呈现。\n\n"}
            })

            result = "已利用 Default 遗留模式的流式事件发射机制处理完毕。"

        # 最终状态更新（两种模式下均能完美保留）
        await __event_emitter__({
            "type": "status",
            "data": {"description": "计算大功告成！", "done": True}
        })

        return result
🔧 常见事件发射器问题定位与排查

排查指南与应对策略：

现象：消息内容在中途闪烁，随后彻底消失

  • 根源：在 Native 模式中非法调用了 message / chat:message / replace 事件 — 它们会被模型生成结束时的 chat 快照强制覆盖。
  • 应对方案：请将原本需要用事件发送的最终内容，作为工具的返回值（return value）进行交付，并在执行中途使用 status 状态栏来提供进度提示。切忌为了图省事而让用户切换到不再维护的遗留 Default 模式中去运行工具。

现象：工具仿佛卡死，对指令完全没有任何反应

  • 根源：模型的工具调用权限没有被正常开启。
  • 应对方案：请前往模型设置页面进行工具赋能，或点击聊天输入框旁边的 + 号开启对该工具的使用授权。

现象：事件完全没有在前端聊天中渲染出来

  • 根源：在工具签名中缺失了 __event_emitter__ 参数，或者该参数为 None。
  • 应对方案：时刻确保您在编写的工具方法签名里正确声明了 __event_emitter__。

现象：引文链接失效或被奇怪的系统引文覆盖

  • 根源：未将 self.citation 设置为 False。
  • 应对方案：在 __init__ 方法中配置 self.citation = False。

诊断工具包 (Diagnostic Tool)：

async def event_diagnostics_tool(
        self, __event_emitter__=None, __metadata__=None, __user__=None
    ) -> str:
        """
        针对事件发射器进行全方位健康体检的综合调试工具
        """

        report = ["# 🔍 事件发射器全项健康诊断报告\n"]

        # 检测事件发射器注入状态
        if __event_emitter__:
            report.append("✅ 事件发射器（Event Emitter）：**正常注入**\n")
        else:
            report.append("❌ 事件发射器（Event Emitter）：**注入失败**\n")
            return "".join(report)

        # 检测聊天元数据及调用模式
        if __metadata__:
            mode = __metadata__.get("params", {}).get("function_calling", "default")
            report.append(f"✅ 当前函数调用模式（Function Calling Mode）: **{mode}**\n")
        else:
            report.append("⚠️ 元数据（Metadata）缺失 (当前调用模式不明)\n")
            mode = "unknown"

        # 检测用户上下文
        if __user__:
            report.append("✅ 用户上下文注入：**正常可用**\n")
        else:
            report.append("⚠️ 用户上下文注入：**缺失**\n")

        # 对完全兼容的事件进行体检（可在所有模式下安全运转）
        report.append("\n## 测试兼容的事件类型：\n")

        try:
            await __event_emitter__({
                "type": "status",
                "data": {"description": "正在诊断 status 事件...", "done": False}
            })
            report.append("✅ 状态栏（Status）事件：**检测合格**\n")
        except Exception as e:
            report.append(f"❌ 状态栏（Status）事件：**出现故障** - {str(e)}\n")

        try:
            await __event_emitter__({
                "type": "notification",
                "data": {"content": "测试 Toast 气泡"}
            })
            report.append("✅ 气泡通知（Notification）事件：**检测合格**\n")
        except Exception as e:
            report.append(f"❌ 气泡通知（Notification）事件：**出现故障** - {str(e)}\n")

        # 对特定模式敏感的易错事件进行体检
        report.append("\n## 测试模式敏感事件：\n")

        try:
            await __event_emitter__({
                "type": "message",
                "data": {"content": "**测试 message 事件发送** — 该条测试仅能在遗留 Default 模式下被成功持久化\n"}
            })
            report.append("✅ 消息（Message）事件：**已发送**（请注意：在 Native 模式下该行可能会被快照抹去）\n")
        except Exception as e:
            report.append(f"❌ 消息（Message）事件：**出现故障** - {str(e)}\n")

        # 终结诊断状态
        await __event_emitter__({
            "type": "status",
            "data": {"description": "体检完毕！", "done": True}
        })

        # 提供诊断建议
        report.append("\n## 诊断整改建议：\n")

        if mode == "native":
            report.append("""
⚠️ **检测到当前运行在 Native 模式下**：此模式仅开放受限的事件通道
- ✅ **推荐使用**：status（状态）、citation（引文）、notification（提示）、files（文件挂载）
- ❌ **严禁依赖**：message、replace、chat:message 等容易被覆盖的流式修改事件
- 💡 若因老工具未重构必须维持原交互体验，可选择将模型切换回 Default 遗留模式（不推荐）
""")
        elif mode == "default":
            report.append("""
✅ **检测到当前运行在 Default 模式下**：此模式支持全套事件发射器
- 所有的事件均可以完美持久地在界面中渲染
- 能够顺畅地开展基于流式消息和页面替换的内容交互
""")
        else:
            report.append("""
❓ **调用模式未知**：请核实模型的开发设置
- 检查该模型是否开启了工具调用参数
- 核对模型的底层服务提供商是否符合工具调用要求
""")

        return "".join(report)
📚 事件发射器速查速记手册

在任何环境下均完全兼容（极力推荐）：

# 状态栏进度更新 — 给用户清晰的处理反馈
await __event_emitter__({
    "type": "status",
    "data": {"description": "正在计算分析中...", "done": False}
})

# 知识库引文与溯源 — 提供严肃的数据追溯支持
await __event_emitter__({
    "type": "citation",
    "data": {
        "document": ["具体的数据内容"],
        "source": {"name": "引用数据源名称", "url": "https://example.com"}
    }
})

# 系统 Toast 弹窗通知
await __event_emitter__({
    "type": "notification",
    "data": {"content": "系统任务成功结束！"}
})

仅能在 Default 遗留模式下工作（在 Native 下会失效）：

# ⚠️ 下述事件在 Native 模式中极易发生内容闪烁并被快照直接抹除！

# 在消息体中逐字追加流式内容
await __event_emitter__({
    "type": "message",
    "data": {"content": "正在流式加载下一段文字..."}
})

# 整体擦除并改写该条消息的全部内容
await __event_emitter__({
    "type": "replace",
    "data": {"content": "这是覆写之后的全新整段内容"}
})

# Delta 差异文本追加
await __event_emitter__({
    "type": "chat:message:delta",
    "data": {"content": "增量的追加字段"}
})

自适应模式检测常用方法：

def get_function_calling_mode(__metadata__):
    """用于快速检测当前聊天处于哪种工具调用模式的通用工具"""
    if not __metadata__:
        return "unknown"
    return __metadata__.get("params", {}).get("function_calling", "default")

# 在工具的运行逻辑中进行检测：
mode = get_function_calling_mode(__metadata__)
is_native = (mode == "native")
can_stream_messages = not is_native

必备的工具头导入声明：

import asyncio
from datetime import datetime
from typing import Optional, Callable, Awaitable

嵌入富 UI 元素​

Tools 和 Actions 可以直接返回一段 HTML 内容，它们会在聊天界面中作为一个个可交互的 iframe 模块进行完全的渲染。有关完整的设计规范、优秀示例、高级安全考量以及跨域 CORS 相关的环境配置参数，请参阅专门的 Rich UI 交互式内联渲染开发指南。

外部第三方库依赖包​

在 Tools 文件顶部的文档字符串元数据中，您可以像下面声明那样，指定自定义的第三方 Python 依赖包。当您在管理员界面点击 Save 时，Open WebUI 会自动解析该行依赖，并在后台立刻通过 pip install 将声明的所有依赖项一次性下载安装完毕。

注意

🚨 极其重要：警惕第三方库版本的命名冲突

当系统内不同的 tools 分别依赖了同一个第三方包的不同版本时（例如，工具 A 要求 pandas==1.5.0，而工具 B 则要求 pandas==2.0.0），Open WebUI 在安装它们时会呈现不确定的先后次序。这极易导致库版本发生覆盖冲突，从而导致原本工作正常的工具因依赖版本不兼容而发生瘫痪。

避免此冲突唯一鲁棒且行之有效的方案是部署 OpenAPI 工具服务器。

我们强烈建议对于复杂或多依赖的业务场景，请使用 OpenAPI 工具服务器 来运行您的工具，以实现完美的依赖进程隔离。

针对多 Worker 或生产环境的严正警告

切勿在生产环境或高并发的多 Worker 部署中依赖运行时的动态 pip 依赖安装！ 当在生产环境中配置了 UVICORN_WORKERS > 1 或者部署了多个容器实例（Replica）时，系统启动时每个 Worker 进程或容器副本都会在后台尝试并发运行 pip 进行依赖项的安装。这会在文件系统层面导致严重的读写竞争（Race Conditions），导致并发执行的 pip 锁死冲突，并由于 pip 的内部文件锁竞争而直接抛出 AssertionError 导致容器瘫痪崩溃。

请务必在生产环境中配置环境变量 ENABLE_PIP_INSTALL_FRONTMATTER_REQUIREMENTS=False，以从底层彻底关闭在运行中动态调用 pip 安装依赖的机制。随后，请在进行容器镜像打包或镜像构建时，通过编写自定义的 Dockerfile，提前将您工具所需要的一切第三方 Python 依赖库打包进镜像内部：

FROM ghcr.io/open-webui/open-webui:main

RUN pip install --no-cache-dir python-docx requests beautifulsoup4

运行时动态安装依赖包的功能仅适用于单 Worker 运行的测试环境，或者供您在个人家用服务器（Homelab）上进行新功能或新 tools 代码研发时的快速迭代。对于任何承载多用户并发访问的严肃生产级部署，请将所有依赖项固化在容器镜像之中。

备注

请务必谨记，运行时动态调用的 pip 进程是在 Open WebUI 本身相同的后端进程中启动并运行的，因此在动态安装依赖包期间，Open WebUI 的前端界面会暂时陷入完全无响应的冻结状态。

此外，自动 pip 动态安装并不会为您解决或规避新安装的库与 Open WebUI 本身既有第三方依赖包的命名与版本冲突。如果在使用中不够小心，随意的 requirements 声明甚至可能会破坏或覆盖 Open WebUI 本身运行所依赖的基础库，导致系统不可用。您可能需要在声明依赖时，将 open-webui 自身也加入要求列表里以尝试规避。

声明依赖的元数据示例
"""
title: myToolName
author: myName
funding_url: [在此处声明的任何链接，都会在聊天界面中转化为一个红色的“爱心”赞助按钮，以方便用户对您表示实质的支持]
version: 1.0.0

# 版本号会清晰呈现在 UI 管理界面中，以方便用户了解工具的更新轨迹。
license: GPLv3
description: [强烈推荐编写简明实用的工具用途描述]
requirements: package1>=2.7.0,package2,package3
"""
This content is for informational purposes only and does not constitute a warranty, guarantee, or contractual commitment. Open WebUI is provided "as is." See your license for applicable terms.