Open WebUI工具系统：自定义工具开发指南

2026-02-05 04:35:30作者：胡唯隽

Open WebUI作为一款功能丰富的自托管WebUI，其工具系统允许用户扩展LLM（大型语言模型）的能力边界。本文将深入剖析工具系统的架构设计，提供从环境搭建到高级功能实现的全流程开发指南，帮助开发者构建满足特定业务需求的自定义工具。

工具系统架构解析

Open WebUI工具系统采用分层架构设计，通过数据模型层、API接口层和执行引擎层实现功能解耦。核心模块位于backend/open_webui目录下，主要包含工具数据模型定义和API路由实现。

核心数据模型

工具系统的数据模型定义在backend/open_webui/models/tools.py中，采用SQLAlchemy ORM框架实现与数据库的交互。核心模型包括：

Tool：存储工具元数据，包括ID、名称、代码内容、规格说明等
ToolMeta：工具元数据扩展，包含描述信息和清单配置
ToolModel：工具数据传输对象(DTO)，用于API交互
ToolsTable：工具数据库操作封装类，提供CRUD操作接口

数据模型关系如图所示：

classDiagram
    class Tool {
        +str id
        +str user_id
        +str name
        +str content
        +list[dict] specs
        +ToolMeta meta
        +dict access_control
        +int updated_at
        +int created_at
    }
    
    class ToolMeta {
        +str description
        +dict manifest
    }
    
    class ToolModel {
        +str id
        +str user_id
        +str name
        +str content
        +list[dict] specs
        +ToolMeta meta
        +dict access_control
        +int updated_at
        +int created_at
    }
    
    class ToolsTable {
        +insert_new_tool()
        +get_tool_by_id()
        +get_tools()
        +update_tool_by_id()
        +delete_tool_by_id()
    }
    
    Tool "1" --> "1" ToolMeta : contains
    ToolModel "1" --> "1" ToolMeta : contains
    ToolsTable "1" --> "*" Tool : manages

API接口设计

工具系统的API接口定义在backend/open_webui/routers/tools.py中，基于FastAPI框架实现RESTful风格的接口设计。主要接口包括：

接口路径	方法	功能描述	权限要求
`/`	GET	获取工具列表	已认证用户
`/create`	POST	创建新工具	管理员或有权限用户
`/id/{id}`	GET	获取工具详情	工具所有者或管理员
`/id/{id}/update`	POST	更新工具	工具所有者或管理员
`/id/{id}/delete`	DELETE	删除工具	工具所有者或管理员
`/id/{id}/valves`	GET	获取工具阀门配置	已认证用户
`/id/{id}/valves/update`	POST	更新工具阀门配置	工具所有者或管理员

工具创建流程通过create_new_tools函数实现，核心步骤包括权限验证、工具ID合法性检查、代码内容处理、工具模块加载和数据库存储。

开发环境搭建

环境准备

在开始自定义工具开发前，需先完成Open WebUI的本地部署。推荐使用Docker Compose方式部署，确保开发环境一致性：

# 克隆代码仓库
git clone https://gitcode.com/GitHub_Trending/op/open-webui
cd open-webui

# 使用Docker Compose启动服务
docker-compose up -d

开发工具链

自定义工具开发需要以下工具链支持：

Python 3.10+：工具代码执行环境
FastAPI：API接口开发框架
Pydantic：数据验证和模型定义
SQLAlchemy：数据库ORM框架
Git：版本控制工具

可以通过以下命令安装开发依赖：

# 安装Python依赖
cd backend
pip install -r requirements.txt

项目结构

工具开发涉及的核心目录结构如下：

open-webui/
├── backend/
│   └── open_webui/
│       ├── models/
│       │   └── tools.py      # 工具数据模型定义
│       ├── routers/
│       │   └── tools.py      # 工具API接口定义
│       ├── utils/
│       │   └── tools.py      # 工具相关工具函数
│       └── internal/
│           └── db.py         # 数据库连接管理
└── src/
    └── lib/
        └── components/
            └── tools/        # 前端工具相关组件

工具开发全流程

1. 工具定义规范

一个标准的Open WebUI工具由三部分组成：元数据定义、功能实现和阀门配置。以下是一个基础工具模板：

"""
name: 天气查询工具
description: 提供实时天气信息查询服务
author: Open WebUI团队
version: 1.0.0
"""

from pydantic import BaseModel, Field
from typing import Optional

# 输入参数模型
class InputSchema(BaseModel):
    city: str = Field(..., description="城市名称")
    date: Optional[str] = Field(None, description="查询日期，格式YYYY-MM-DD")

# 输出结果模型
class OutputSchema(BaseModel):
    city: str
    date: str
    temperature: str
    weather: str
    wind: str

# 工具阀门配置（可选）
class Valves(BaseModel):
    api_key: str = Field(..., description="天气API密钥")
    timeout: int = Field(5, description="请求超时时间(秒)")

# 用户阀门配置（可选）
class UserValves(BaseModel):
    unit: str = Field("celsius", description="温度单位，celsius或fahrenheit")

# 工具主函数
def run(input: InputSchema, valves: Valves, user_valves: UserValves) -> OutputSchema:
    """
    查询指定城市的天气信息
    """
    # 这里实现具体的天气查询逻辑
    # 1. 调用外部天气API
    # 2. 处理返回结果
    # 3. 格式化输出
    
    return OutputSchema(
        city=input.city,
        date=input.date or "2023-10-01",
        temperature="25°C",
        weather="晴朗",
        wind="微风"
    )

工具代码需遵循以下规范：

元数据部分：使用多行注释定义，包含name、description、author等信息
输入/输出模型：使用Pydantic的BaseModel定义，支持字段描述和验证
阀门配置：Valves类定义全局配置，UserValves定义用户级配置
主函数：run函数作为入口点，接收输入参数、阀门配置，返回输出结果

2. 工具创建实现

工具创建通过调用POST /tools/create接口实现，核心逻辑在create_new_tools函数中。以下是创建工具的关键代码流程：

@router.post("/create", response_model=Optional[ToolResponse])
async def create_new_tools(
    request: Request,
    form_data: ToolForm,
    user=Depends(get_verified_user),
):
    # 1. 权限验证
    if user.role != "admin" and not has_permission(...):
        raise HTTPException(...)
    
    # 2. ID合法性检查
    if not form_data.id.isidentifier():
        raise HTTPException(...)
    
    # 3. 检查工具ID是否已存在
    tools = Tools.get_tool_by_id(form_data.id)
    if tools is None:
        try:
            # 4. 处理工具代码内容
            form_data.content = replace_imports(form_data.content)
            
            # 5. 加载工具模块并解析规格
            tools_module, frontmatter = load_tools_module_by_id(
                form_data.id, content=form_data.content
            )
            form_data.meta.manifest = frontmatter
            
            # 6. 获取工具规格
            specs = get_tools_specs(TOOLS[form_data.id])
            
            # 7. 保存工具到数据库
            tools = Tools.insert_new_tool(user.id, form_data, specs)
            
            return tools
        except Exception as e:
            raise HTTPException(...)
    else:
        raise HTTPException(...)

工具创建流程如图所示：

flowchart TD
    A[开始] --> B[权限验证]
    B --> C{验证通过?}
    C -->|否| D[返回401错误]
    C -->|是| E[ID格式检查]
    E --> F{格式合法?}
    F -->|否| G[返回400错误]
    F -->|是| H[检查ID是否存在]
    H --> I{已存在?}
    I -->|是| J[返回400错误]
    I -->|否| K[处理工具代码]
    K --> L[加载工具模块]
    L --> M[解析工具规格]
    M --> N[保存到数据库]
    N --> O[返回创建结果]
    O --> P[结束]

3. 工具执行机制

工具执行通过get_tools_specs函数解析工具规格，然后在LLM对话过程中动态调用。工具调用流程如下：

LLM根据用户问题判断是否需要调用工具
若需要调用工具，根据工具规格生成参数
调用工具的run函数执行具体逻辑
获取工具返回结果，整理后返回给LLM
LLM基于工具结果生成最终回答

工具执行的核心代码在utils/tools.py中的get_tools_specs函数，该函数解析工具模块，提取输入/输出模型、描述信息等，生成工具规格。

核心功能实现

权限控制

工具系统实现了细粒度的权限控制，基于用户角色和访问控制列表(ACL)实现。权限控制逻辑在backend/open_webui/utils/access_control.py中实现。

工具访问控制通过access_control字段实现，支持三种权限模式：

公开访问：access_control为None时，所有用户可访问
私有访问：access_control为{}时，仅工具所有者可访问
自定义访问：指定可读写的用户ID或组ID列表

# 自定义访问控制示例
{
  "read": {
    "group_ids": ["group_id1", "group_id2"],
    "user_ids": ["user_id1", "user_id2"]
  },
  "write": {
    "group_ids": ["group_id1"],
    "user_ids": ["user_id1"]
  }
}

权限检查通过has_access函数实现，在工具查询、更新、删除等操作前进行验证。

阀门系统

阀门系统(Valves)是Open WebUI工具的特色功能，用于管理工具的配置参数和访问控制。阀门分为两种类型：

Tool Valves：工具级阀门，存储全局配置，如API密钥、超时设置等
User Valves：用户级阀门，存储用户个性化配置，如温度单位、偏好设置等

阀门定义通过Pydantic模型实现，支持自动验证和前端表单生成。阀门配置存储在数据库中，通过get_tool_valves_by_id和update_tool_valves_by_id函数进行管理。

阀门使用流程：

在工具代码中定义Valves和UserValves模型
系统自动生成配置表单
管理员或用户填写配置值
工具执行时自动注入阀门配置

代码安全机制

为确保工具代码的安全性，系统实现了多项安全机制：

代码沙箱：工具代码在受限环境中执行，限制危险操作
导入替换：通过replace_imports函数过滤危险模块导入
超时控制：工具执行设置超时时间，防止无限循环
资源限制：限制工具的CPU、内存使用，防止资源滥用

代码导入替换逻辑示例：

def replace_imports(content: str) -> str:
    """过滤危险模块导入"""
    forbidden_imports = {
        "os", "subprocess", "sys", "shutil", "tempfile",
        "pickle", "eval", "exec", "compile"
    }
    
    for imp in forbidden_imports:
        content = re.sub(
            rf"import\s+{imp}\b",
            f"# import {imp} (forbidden)",
            content
        )
        content = re.sub(
            rf"from\s+{imp}\s+import",
            f"# from {imp} import (forbidden)",
            content
        )
    
    return content

实战案例：天气查询工具

工具代码实现

以下是一个完整的天气查询工具实现，使用高德地图API获取天气数据：

"""
name: 高德天气查询
description: 使用高德地图API查询实时天气信息
author: 开发者名称
version: 1.0.0
"""

import requests
from pydantic import BaseModel, Field
from typing import Optional

# 输入参数模型
class InputSchema(BaseModel):
    city: str = Field(..., description="城市名称，如：北京")
    date: Optional[str] = Field(None, description="查询日期，格式YYYY-MM-DD，默认今天")

# 输出结果模型
class OutputSchema(BaseModel):
    city: str = Field(..., description="城市名称")
    date: str = Field(..., description="日期")
    weather: str = Field(..., description="天气状况")
    temperature: str = Field(..., description="温度")
    humidity: str = Field(..., description="湿度")
    wind: str = Field(..., description="风向风力")
    report_time: str = Field(..., description="数据更新时间")

# 工具阀门配置
class Valves(BaseModel):
    api_key: str = Field(..., description="高德地图API密钥")
    timeout: int = Field(5, description="请求超时时间(秒)")

# 用户阀门配置
class UserValves(BaseModel):
    temperature_unit: str = Field("celsius", description="温度单位，celsius(摄氏度)或fahrenheit(华氏度)")

def run(input: InputSchema, valves: Valves, user_valves: UserValves) -> OutputSchema:
    """
    使用高德地图API查询天气信息
    """
    # 1. 获取城市编码
    city_url = f"https://restapi.amap.com/v3/config/district?keywords={input.city}&key={valves.api_key}"
    city_response = requests.get(city_url, timeout=valves.timeout)
    city_data = city_response.json()
    
    if city_data["status"] != "1" or not city_data["districts"]:
        raise ValueError(f"无法找到城市: {input.city}")
    
    city_code = city_data["districts"][0]["adcode"]
    
    # 2. 查询天气信息
    weather_url = f"https://restapi.amap.com/v3/weather/weatherInfo?city={city_code}&key={valves.api_key}"
    weather_response = requests.get(weather_url, timeout=valves.timeout)
    weather_data = weather_response.json()
    
    if weather_data["status"] != "1" or not weather_data["lives"]:
        raise ValueError("无法获取天气数据")
    
    live_data = weather_data["lives"][0]
    
    # 3. 处理温度单位
    temperature = live_data["temperature"]
    if user_valves.temperature_unit == "fahrenheit":
        # 摄氏度转华氏度
        temperature = f"{int(temperature) * 9 // 5 + 32}°F"
    else:
        temperature = f"{temperature}°C"
    
    # 4. 构造返回结果
    return OutputSchema(
        city=live_data["city"],
        date=input.date or live_data["reporttime"].split()[0],
        weather=live_data["weather"],
        temperature=temperature,
        humidity=f"{live_data['humidity']}%",
        wind=f"{live_data['winddirection']} {live_data['windpower']}级",
        report_time=live_data["reporttime"]
    )

工具部署与使用

部署步骤

登录Open WebUI管理界面
导航到"工具"页面，点击"创建工具"
填写工具ID：amap_weather
填写工具名称：高德天气查询
将上述代码粘贴到内容框中
点击"创建"按钮提交

配置阀门

在工具列表中找到"高德天气查询"，点击"配置"
填写高德地图API密钥（需提前在高德开放平台申请）
设置超时时间：5秒
点击"保存配置"

使用方法

在聊天界面输入：查询北京今天的天气，系统会自动调用天气查询工具，返回结果如下：

北京今天的天气情况：
- 日期：2023-10-01
- 天气：晴朗
- 温度：25°C
- 湿度：35%
- 风向风力：南风 2级
- 数据更新时间：2023-10-01 08:00

高级功能

多模态工具开发

Open WebUI支持开发处理图像、音频等多模态数据的工具。以下是一个图像分析工具的示例框架：

"""
name: 图像分析工具
description: 分析图像内容并生成描述
author: Open WebUI团队
version: 1.0.0
"""

from pydantic import BaseModel, Field
from typing import Optional
from PIL import Image
import base64
from io import BytesIO

class InputSchema(BaseModel):
    image_data: str = Field(..., description="Base64编码的图像数据")
    detail_level: str = Field("medium", description="分析详细程度：low, medium, high")

class OutputSchema(BaseModel):
    description: str = Field(..., description="图像描述")
    objects: list[str] = Field(..., description="识别到的物体列表")
    colors: list[str] = Field(..., description="主要颜色列表")

class Valves(BaseModel):
    model_name: str = Field("clip-vit-base-patch32", description="图像模型名称")
    max_objects: int = Field(10, description="最大识别物体数量")

def run(input: InputSchema, valves: Valves) -> OutputSchema:
    # 1. 解码Base64图像数据
    image_data = base64.b64decode(input.image_data)
    image = Image.open(BytesIO(image_data))
    
    # 2. 调用图像分析模型
    # 这里可以集成CLIP、YOLO等模型
    
    # 3. 处理分析结果
    return OutputSchema(
        description="一只猫坐在沙发上",
        objects=["猫", "沙发", "地毯"],
        colors=["灰色", "蓝色", "白色"]
    )

工具链与工作流

复杂任务可以通过工具链(Workflow)将多个工具组合使用。工具链通过定义工具执行顺序和数据流转实现：

"""
name: 旅行规划工具链
description: 规划旅行行程，集成天气查询、景点推荐和酒店预订
author: Open WebUI团队
version: 1.0.0
"""

from pydantic import BaseModel, Field
from typing import List, Optional

# 输入模型
class InputSchema(BaseModel):
    destination: str = Field(..., description="目的地城市")
    start_date: str = Field(..., description="开始日期，格式YYYY-MM-DD")
    end_date: str = Field(..., description="结束日期，格式YYYY-MM-DD")
    travelers: int = Field(1, description="旅行人数")

# 输出模型
class OutputSchema(BaseModel):
    destination: str
    date_range: str
    weather_forecast: dict
    attractions: List[dict]
    hotel_recommendations: List[dict]
    itinerary: List[dict]

def run(input: InputSchema, workflow: dict) -> OutputSchema:
    """
    旅行规划工作流，依次调用天气查询、景点推荐和酒店预订工具
    """
    # 1. 查询天气
    weather_result = workflow["tools"]["weather"].run({
        "city": input.destination,
        "start_date": input.start_date,
        "end_date": input.end_date
    })
    
    # 2. 推荐景点
    attractions_result = workflow["tools"]["attractions"].run({
        "city": input.destination,
        "interest": "tourist",
        "duration": (end_date - start_date).days
    })
    
    # 3. 酒店预订
    hotel_result = workflow["tools"]["hotel"].run({
        "city": input.destination,
        "check_in": input.start_date,
        "check_out": input.end_date,
        "guests": input.travelers,
        "budget": "medium"
    })
    
    # 4. 生成行程计划
    itinerary = generate_itinerary(
        weather_result, attractions_result, hotel_result, input
    )
    
    return OutputSchema(
        destination=input.destination,
        date_range=f"{input.start_date} - {input.end_date}",
        weather_forecast=weather_result,
        attractions=attractions_result["attractions"],
        hotel_recommendations=hotel_result["hotels"],
        itinerary=itinerary
    )

调试与测试

本地调试

工具开发过程中可以通过本地脚本进行调试，无需每次都部署到Open WebUI中：

# tool_debugger.py
from pydantic import BaseModel
from your_tool import run, InputSchema, Valves, UserValves

# 模拟输入数据
input_data = InputSchema(
    city="北京",
    date="2023-10-01"
)

# 模拟阀门配置
valves = Valves(
    api_key="your_api_key_here",
    timeout=5
)

user_valves = UserValves(
    unit="celsius"
)

# 执行工具函数
try:
    result = run(input_data, valves, user_valves)
    print("工具执行结果:")
    print(result.model_dump_json(indent=2))
except Exception as e:
    print(f"工具执行出错: {e}")

单元测试

为确保工具质量，建议编写单元测试：

# test_weather_tool.py
import pytest
from pydantic import ValidationError
from your_tool import run, InputSchema, Valves, UserValves

def test_input_validation():
    # 测试输入验证
    with pytest.raises(ValidationError):
        InputSchema()  # 缺少必填字段city

def test_successful_run():
    # 测试正常执行流程
    input_data = InputSchema(city="北京")
    valves = Valves(api_key="test_key", timeout=5)
    user_valves = UserValves(unit="celsius")
    
    result = run(input_data, valves, user_valves)
    assert result.city == "北京"
    assert "temperature" in result.model_dump()

def test_invalid_api_key():
    # 测试API密钥无效的情况
    input_data = InputSchema(city="北京")
    valves = Valves(api_key="invalid_key", timeout=5)
    user_valves = UserValves(unit="celsius")
    
    with pytest.raises(Exception):
        run(input_data, valves, user_valves)

日志与监控

Open WebUI提供了完善的日志系统，工具执行过程中的日志可以通过以下方式查看：

# 查看容器日志
docker logs open-webui-backend -f

# 过滤工具相关日志
docker logs open-webui-backend -f | grep "tools"

工具执行日志定义在backend/open_webui/models/tools.py中，使用Python的logging模块实现：

import logging

log = logging.getLogger(__name__)
log.setLevel(SRC_LOG_LEVELS["MODELS"])

# 在关键位置添加日志
log.info(f"Creating tool: {form_data.id} by user {user.id}")
log.error(f"Error creating tool: {e}")