OpenWebUI监听ipv6教程，部署mcpo并添加和使用工具

zhang0281 · 2025 年6 月 21 日 15:13

从上一帖使用docker compose ，搭建 OpenWebUI 教程，部分配置优化，启用ipv6 - 开发调优 - LINUX DO 继续讨论。

配置文件

需要手动替换<your-secret-key>，替换为随机字符串即可。替换后示例：WEBUI_SECRET_KEY: AAAABBBB。WEBUI_SECRET_KEY可以使重启容器不丢失登录状态。

docker-compose.yml


version: "3.7"

x-defaults: &defaults
  restart: unless-stopped
  networks:
    - open-webui
  extra_hosts:
    - host.docker.internal:host-gateway

x-watchtower: &watchtower
  labels:
    - com.centurylinklabs.watchtower.enable=true

x-logging: &default-logging
  logging:
    driver: "json-file"
    options:
      max-size: "10m"
      max-file: "3"


services:
  tts:
    image: mzzsfy/tts:latest
    container_name: tts
    <<: [*defaults, *watchtower, *default-logging]
    environment:
      TTS_STRICT_MODE: true

  new-api:
    image: calciumion/new-api:latest
    container_name: new-api
    <<: [*defaults, *watchtower, *default-logging]
    command: --log-dir /app/logs
    ports:
      - "3001:3000"
    environment:
      SQL_DSN: postgres://openwebui:mypasswd@postgres:5432/new-api
      REDIS_CONN_STRING: redis://redis/0
      SESSION_SECRET: server01
      TZ: Asia/Shanghai
    volumes:
      - ./new-api/data:/data
      - ./new-api/logs:/app/logs
    depends_on:
      redis:
        condition: service_healthy
      tts:
        condition: service_started
    healthcheck:
      test: ["CMD-SHELL", "wget -q -O - http://localhost:3000/api/status | grep -o '\"success\":\\s*true' | awk -F: '{print $2}'"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 15s

  redis:
    image: redis:7-alpine
    container_name: redis
    <<: [*defaults, *default-logging]
    volumes:
      - ./redis/data:/data
      - ./redis/redis.conf:/usr/local/etc/redis/redis.conf
    command: redis-server /usr/local/etc/redis/redis.conf
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3

  postgres:
    container_name: postgres
    image: postgres:17-alpine
    <<: [*defaults, *default-logging]
    environment:
      POSTGRES_USER: openwebui
      POSTGRES_PASSWORD: mypasswd
      POSTGRES_DB: openwebui
      POSTGRES_MULTIPLE_DATABASES: \"new-api\",openwebui,openwebui_monitor
    volumes:
      - ./postgres_data:/var/lib/postgresql/data
      - ./postgres-init:/docker-entrypoint-initdb.d
    ports:
      - "5432:5432"
    #user: postgres
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U openwebui | grep -o 'accepting connections'"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 30s

  open-webui:
    container_name: open-webui
    image: ghcr.io/open-webui/open-webui:main
    <<: [*defaults, *default-logging]
    ports:
      - "4567:8080"
    volumes:
      - ./open-webui_data:/app/backend/data
      - ./open-webui-assets-custom:/app/build/assets/custom
      - ./open-webui-init.sh:/app/backend/open-webui-init.sh
    environment:
      OLLAMA_BASE_URL: false
      SEARXNG_URL: false
      ENABLE_REALTIME_CHAT_SAVE: false
      DATABASE_URL: postgres://openwebui:mypasswd@postgres:5432/openwebui
      DATABASE_POOL_SIZE: 10
      DATABASE_POOL_MAX_OVERFLOW: 20
      REDIS_URL: redis://redis/1
      ENABLE_WEBSOCKET_SUPPORT: True
      WEBSOCKET_MANAGER: redis
      ENABLE_OLLAMA_API: false
      AIOHTTP_CLIENT_TIMEOUT_OPENAI_MODEL_LIST: 5
      #AIOHTTP_CLIENT_TIMEOUT: 10
      HF_ENDPOINT: https://hf-mirror.com
      WEBUI_SECRET_KEY: <your-secret-key>
      HOST: "0.0.0.0"
    depends_on:
      postgres:
        condition: service_healthy
      new-api:
        condition: service_healthy
 #     pipelines:
 #       condition: service_healthy
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:8080/health | grep -o '\"status\":\\s*true' | awk -F: '{print $2}'"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 30s
    command:
      - /bin/bash
      - -c
      - |
        bash /app/backend/open-webui-init.sh

  openwebui-monitor:
    container_name: openwebui-monitor
    image: docker.io/variantconst/openwebui-monitor:latest
    <<: [*defaults, *watchtower, *default-logging]
    ports:
      - "7878:3000"
    env_file:
      - .openwebui-monitor-env
    environment:
      POSTGRES_HOST: postgres
      POSTGRES_PORT: 5432
      POSTGRES_USER: openwebui
      POSTGRES_PASSWORD: mypasswd
      POSTGRES_DATABASE: openwebui_monitor
    depends_on:
      postgres:
        condition: service_healthy

  mcpo:
    container_name: mcpo
    <<: [*defaults, *watchtower, *default-logging]
    image: ghcr.io/open-webui/mcpo:main
    ports:
      - "8001:8000"
    volumes:
      - ./mcpo/config.json:/app/config.json
      - ./mcpo/mcp:/home/mcp
    command: --config config.json

networks:
  open-webui:
    driver: bridge
    enable_ipv6: true
    ipam:
      driver: default
      config:
        - subnet: fd00:db8:1234::/64
          gateway: fd00:db8:1234::1

距上一帖变化：

所有 environment 修改格式为键值对

涉及到redis的服务单独分配了数据库

redis添加了配置文件以实现持久化

openwebui使用了另一种方式实现双栈监听

添加了mcpo服务以使用mcp

./postgres-init/create-multiple-db.sh无变化，如需要请到上一帖查看。

./open-webui-assets-custom/custom.js无变化，如需要请到上一帖查看。

./open-webui-assets-custom/custom.css

@font-face { font-family: 'Dank Mono'; src:
    url('../assets/fonts/DankMono-Regular.woff2')
    format('woff2'); font-display: swap; font-style:
    normal;
}
@font-face { font-family: 'Dank Mono'; src:
    url('../assets/fonts/DankMono-Italic.woff2')
    format('woff2'); font-display: swap; font-style:
    italic;
}
@font-face { font-family: 'PingFangSC-Regular'; src:
    url('../assets/fonts/PingFangSC-Regular.woff2')
    format('woff2'); font-display: swap;
}
html { scroll-behavior: smooth;
}
body { font-family: 'PingFangSC-Regular', -apple-system,
    BlinkMacSystemFont, sans-serif;
}
/* 代码块容器样式 */ .language-javascript,
[class*="language-"] {
    background: #f6f8fa !important; border-radius: 10px
    !important;
    box-shadow: 0 10px 30px 0 rgba(0, 0, 0, 0.1)
    !important;
    position: relative; margin: 1.2em 0;
}
.dark .language-javascript, .dark [class*="language-"] {
    background: #282c34 !important; box-shadow: 0 10px
    30px 0 rgba(0, 0, 0, .4) !important;
}
/* 代码块顶部栏 */ .sticky.top-8 { background: #e1e4e8
    !important;
    height: 40px !important; display: flex; align-items:
    center; border-radius: 10px 10px 0 0; padding: 0 15px
    !important;
    margin-bottom: -59px !important;
}
.dark .sticky.top-8 { background: #21252b !important;
}
/* 语言标识 */ .text-text-300 { position: absolute;
    left: 60px; top: 2px; color: #abb2bf !important;
    font-size: 17px !important; font-weight: 500
    !important;
    z-index: 11;
}
.dark .text-text-300 { color: #586069 !important;
}
/* 顶部按钮样式优化 */ .save-code-button,
.copy-code-button, .run-code-button {
    background: #f8f9fa !important; color: #333
    !important;
    border: 1px solid #d1d5da !important; font-size: 12px
    !important;
    padding: 4px 12px !important; border-radius: 4px
    !important;
    transition: all 0.2s ease-in-out !important;
}
.dark .save-code-button, .dark .copy-code-button, .dark
.run-code-button {
    background: #323842 !important; color: #abb2bf
    !important;
    border: 1px solid #3e4451 !important;
}
.save-code-button:hover, .copy-code-button:hover {
    background: #e9ecef !important; color: #222
    !important;
}
.dark .save-code-button:hover, .dark
.copy-code-button:hover {
    background: #3e4451 !important; color: #fff
    !important;
}
/* 代码块顶部装饰圆点 */ .language-javascript::before,
[class*="language-"]::before {
    content: " "; position: absolute; border-radius: 50%;
    background: #ff5f56; width: 12px; height: 12px; left:
    15px; top: 14px; box-shadow: 20px 0 #ffbd2e, 40px 0
    #27c93f;
    z-index: 10;
}
.dakr .language-javascript::before, .dakr
[class*="language-"]::before {
    background: #fc625d; box-shadow: 20px 0 #fdbc40, 40px
    0 #35cd4b;
}
/* 代码内容区域 */ .cm-content { font-family: 'Dank
    Mono', -apple-system, BlinkMacSystemFont, Inter,
    ui-sans-serif, system-ui, 'Segoe UI', Roboto, Ubuntu,
    Cantarell, 'Noto Sans', sans-serif, 'Helvetica Neue',
    Arial, 'Apple Color Emoji', 'Segoe UI Emoji', 'Segoe
    UI Symbol', 'Noto Color Emoji'; font-size: 15px
    !important;
    line-height: 1.6em !important; padding: 20px 1.4em 1em
    30px !important; color: #24292e !important;
}
.dark .cm-content { color: #abb2bf !important;
}
/* 代码语法高亮 */ .cm-line .ͼb { /* 关键字 */ color:
    #d73a49 !important;
}
.dark .cm-line .ͼb { color: #c678dd !important;
}
.cm-line .ͼd { /* 数字 */ color: #a29bfe !important;
}
.dakr .cm-line .ͼd { color: #e5c07b !important;
}
.cm-line .ͼe { /* 字符串 */ color: #6a89cc !important;
}
.dakr .cm-line .ͼe { color: #98c379 !important;
}
.cm-line .ͼg { /* 变量 */ color: #2ca9e1 !important;
    font-style: italic;
}
.dakr .cm-line .ͼg { color: #e3adb9 !important;
}
/* 代码语法高亮 - 扩展 */ .cm-comment { /* 注释 */
    color: #7f848e !important; font-style: italic;
}
.cm-property { color: #61afef !important;
}
cm-tag { color: #e06c75 !important;
}
.cm-attribute { color: #d19a66 !important;
}
.cm-string { color: #98c379 !important;
}
.cm-operator { color: #56b6c2 !important;
}
span.ͼc { color: #7d5fff !important;
}
span.ͼl { color: #6bddcd !important;
}
span.ͼt { /* 暗色模式下的逗号 */ color: #ddb078
    !important;
    ;
    font-style: italic;
}
span.ͼr { /* 暗色模式下的函数名 */ font-style: italic;
}
span.ͼf { /* 亮色模式下奇怪的符号 */ color: #70a1ff;
}
span.ͼm { /* 亮色模式下的注释 */ color: #f29a76;
    font-style: italic;
}
span.ͼw { /* 暗色模式下的注释 */ font-style: italic;
}
/* 滚动条样式 */ .cm-scroller::-webkit-scrollbar {
    height: 10px !important; width: 10px !important;
    background-color: #f6f8fa !important;
}
.dark .cm-scroller::-webkit-scrollbar { background-color:
    #282c34 !important;
}
.cm-scroller::-webkit-scrollbar-track { box-shadow: inset
    0 0 6px rgba(0, 0, 0, 0.1) !important; border-radius:
    10px !important; background-color: #f6f8fa !important;
}
.dark .cm-scroller::-webkit-scrollbar-track { box-shadow:
    inset 0 0 6px rgba(0, 0, 0, .3) !important;
    background-color: #282c34 !important;
}
.cm-scroller::-webkit-scrollbar-thumb { border-radius:
    10px !important; box-shadow: inset 0 0 6px rgba(0, 0,
    0, .2) !important; background-color: #d1d5da
    !important;
}
.dark .cm-scroller::-webkit-scrollbar-thumb { box-shadow:
    inset 0 0 6px rgba(0, 0, 0, .5) !important;
    background-color: #3e4451 !important;
}
/* 行号栏样式 */ .cm-gutters { background: #f6f8fa
    !important;
    border-right: 1px solid #d1d5da !important; color:
    #586069 !important;
    padding-right: 10px !important; font-family: "Dank
    Mono";
}
.dark .cm-gutters { background: #282c34 !important;
    border-right: 1px solid #3e4451 !important; color:
    #495162 !important;
}
/* 当前行高亮 */ .cm-activeLine { background: #6699ff0b
    !important;
}
.cm-gutterElement.cm-activeLineGutter { background-color:
    #f9d3e3;
}
.dark .cm-gutterElement.cm-activeLineGutter {
    background-color: #dd7694;
}
/* 添加代码选中样式 */ .cm-selectionBackground,
.cm-content ::selection {
    background-color: rgba(122, 129, 255, 0.2) !important;
}
.cm-line.cm-selected { background-color: rgba(122, 129,
    255, 0.2) !important;
}
/* 选中时的文本颜色保持原样，确保可读性 */ .cm-content
::selection {
    color: rgba(62, 158, 111, 0.9) !important;
}
.dark .cm-content ::selection { color: rgba(245, 177, 255,
    0.9) !important;
}
/* 匹配相同结果时的颜色 */ .cm-selectionMatch {
    background-color: #9c88ff5a !important;
}
/* 当有多行选中时的样式 */
.cm-selectionLayer>.cm-selectionBackground {
    background-color: rgba(122, 129, 255, 0.2) !important;
}
/* 代码块折叠/展开样式添加与修改 */ .cm-scroller {
    background-color: #f6f8fa;
}
#collapsed>.cm-scroller, #expanded>.cm-scroller {
    padding-bottom: 40px;
}
.dark .cm-scroller { background-color: #282c34;
}
.cm-scroller { overflow: auto !important;
}
.cm-editor { transition: height 1s cubic-bezier(0.4, 0,
    0.2, 1); overflow: hidden !important;
}
/* 只给超高的代码块添加最大高度和内边距 */
.cm-editor#collapsed {
    height: 400px;
}
/* .cm-editor#expanded { padding-bottom: 40px; } */
.code-expand-btn {
    position: absolute; bottom: 10px; left: 50%;
    transform: translateX(-50%); display: flex;
    justify-content: center; align-items: center; padding:
    6px 15px; border-radius: 15px; font-size: 12px;
    cursor: pointer; border: none; color: #666;
    background: rgba(255, 255, 255, 0.6); backdrop-filter:
    blur(8px); box-shadow: 0 4px 6px rgba(0, 0, 0, 0.1);
    -webkit-backdrop-filter: blur(8px); z-index: 11;
    transition: all 0.3s ease;
}
.dark .code-expand-btn { background: rgba(45, 45, 45,
    0.6); color: #fff;
}
.code-expand-btn:hover { background: rgba(255, 255, 255,
    0.8); backdrop-filter: blur(12px);
    -webkit-backdrop-filter: blur(12px); transform:
    translateX(-50%) translateY(-2px); box-shadow: 0 6px
    8px rgba(0, 0, 0, 0.15);
}
.dark .code-expand-btn:hover { background: rgba(45, 45,
    45, 0.8);
}
.code-expand-btn:active { transform: translateX(-50%)
    translateY(0); box-shadow: 0 2px 4px rgba(0, 0, 0,
    0.1);
}
.code-expand-btn::before { content: "⌄"; display:
    inline-block; margin-right: 4px; font-size: 14px;
    transition: transform 0.3s ease;
}
.code-expand-btn#expanded::before { transform:
    rotate(180deg);
}
.code-expand-btn::after { content: "展开代码";
}
.code-expand-btn#expanded::after { content: "收起代码";
}
/* 渐变遮罩 */ .cm-editor#collapsed::after { content:
    ''; position: absolute; bottom: 0; left: 0; right: 0;
    height: 100px; background: linear-gradient(transparent
    0%, rgba(255, 255, 255, 0.3) 40%, rgba(255, 255, 255,
    0.6) 80%, rgba(255, 255, 255, 0.8) 100%);
    pointer-events: none; opacity: 0; transition: opacity
    0.3s ease; z-index: 10; /* 确保遮罩层覆盖到滚动条 */
    width: calc(100% + 17px); /* 17px是标准滚动条宽度 */
}
.dark .cm-editor#collapsed::after { background:
    linear-gradient(transparent 0%, rgba(45, 45, 45, 0.3)
    40%, rgba(45, 45, 45, 0.6) 80%, rgba(45, 45, 45, 0.8)
    100%);
}
/* 只在折叠状态显示渐变遮罩 */ .cm-editor#collapsed::after
{
    opacity: 1;
}
/* 隐藏原始的折叠按钮 */
div.flex.items-center.gap-0\.5 button.flex.gap-1.items-center:not(.run-code-button) {
    display: none;
}
body > div > div:nth-child(2) > div > div {
    flex-wrap: wrap;
}

距上一帖变化：

添加最后一项以使得，在对话页左上角，选择模型的分类，可以显示全部分类。

./open-webui-init.sh

apt-get update && \
    apt-get install -y --no-install-recommends socat && \
    rm -rf /var/lib/apt/lists/*

sed -i '$d' /app/backend/start.sh

cat <<EOF>> /app/backend/start.sh
# 设置外部监听端口和内部应用端口
EXTERNAL_PORT="\${PORT:-8080}"
INTERNAL_PORT=8081 # Uvicorn 将监听这个内部端口
INTERNAL_HOST="127.0.0.1"

# 启动 socat 在后台，创建双栈监听器
# 它会将外部 :8080 的流量转发到内部 127.0.0.1:8081
echo "Starting socat to listen on [::]:\${EXTERNAL_PORT} and forward to \${INTERNAL_HOST}:\${INTERNAL_PORT}"
socat TCP6-LISTEN:\${EXTERNAL_PORT},fork,reuseaddr,ipv6only=0 TCP4:\${INTERNAL_HOST}:\${INTERNAL_PORT} &

# 启动 Uvicorn，但只监听在本地 IPv4 地址上
echo "Starting Uvicorn on \${INTERNAL_HOST}:\${INTERNAL_PORT}"
WEBUI_SECRET_KEY="\$WEBUI_SECRET_KEY" exec "\$PYTHON_CMD" -m uvicorn open_webui.main:app --host "\$INTERNAL_HOST" --port "\$INTERNAL_PORT" --forwarded-allow-ips '*' --workers "\${UVICORN_WORKERS:-1}"
EOF


#cat /app/backend/start.sh
cp /app/build/assets/custom/fonts/* /app/build/assets/fonts/
cp /app/build/assets/custom/custom.css /app/build/assets/
cp /app/build/assets/custom/custom.js /app/build/assets/
sed -i 's|</head>|<link rel="stylesheet" href="/assets/custom.css"></head>|' /app/build/index.html
sed -i 's|</body>|<script src="/assets/custom.js"></script></body>|' /app/build/index.html
bash /app/backend/start.sh

距上一帖变化：

将双栈监听的实现方法从启动两个 uvicorn 分别监听 ipv4 和 ipv6 。修改为，启动一个socat监听容器内的双栈的8080，转发到ipv4的8081端口，并修改uvicorn监听127.0.0.1:8081。

以下是新增的相关文件：

./redis/redis.conf

# 启用 RDB 快照
# 900秒内至少1次修改触发快照
save 900 1
# 300秒内至少10次修改
save 300 10
# 60秒内至少10000次修改
save 60 10000
# RDB 文件名
dbfilename dump.rdb
# 数据存储目录
dir /data

# 启用 AOF 并开启混合持久化
appendonly yes
# 启用 AOF
appendfsync everysec
# 同步频率（每秒同步一次，折中性能与安全）
aof-use-rdb-preamble yes
# 启用混合持久化（Redis 4.0+）

说明：

这个配置文件的主要功能是开启redis的持久化。

./mcpo/config.json

{
  "mcpServers": {
    "amap-maps": {
      "command": "npx",
      "args": [
        "-y",
        "@amap/amap-maps-mcp-server"
      ],
      "env": {
        "AMAP_MAPS_API_KEY": "YOUR_AMAP_MAPS_API_KEY"
      }
    },
    "time": {
      "command": "uvx",
      "args": [
        "mcp-server-time",
        "--local-timezone=Asia/Shanghai"
      ]
    },
    "tavily-mcp": {
      "command": "npx",
      "args": [
        "-y",
        "tavily-mcp"
      ],
      "env": {
        "TAVILY_API_KEY": "YOUR_TAVILY_API_KEY"
      }
    },
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "YOUR_GITHUB_PERSONAL_ACCESS_TOKEN"
      }
    },
    "edgeone-pages-mcp-server": {
      "command": "npx",
      "args": [
        "edgeone-pages-mcp"
      ]
    }
  }
}

说明：

YOUR_AMAP_MAPS_API_KEY 需要被替换为自行申请的高德地图 API KEY。创建应用和Key

YOUR_TAVILY_API_KEY 需要被替换为自行申请的tavily API KEY。免费版每月1000次。tavily KEY申请

YOUR_GITHUB_PERSONAL_ACCESS_TOKEN 需要被替换为你自己的AccessToken。新建AccessToken

OpenWebUI 的设置

管理员面板 → 设置 → 工具 → 添加一个连接(右边加号图标)(左键三击可快速选中下方表格中的内容)

URL	名称	描述
`http://mcpo:8000/amap-maps`	`高德地图`	`地理编码（地址转坐标/坐标转地址）、IP定位、天气查询、POI搜索（关键词/周边）、路径规划（步行/驾车/骑行/公交）、距离测量、以及POI详细信息查询。`
`http://mcpo:8000/time`	`时间`	`获取指定时区的当前时间，以及将时间从一个时区转换到另一个时区。`
`http://mcpo:8000/tavily-mcp`	`网页搜索`	`强大的网页搜索工具（提供实时结果）、内容提取工具（从指定URL获取数据）、网页爬虫（结构化抓取网站内容）和网站地图工具（分析网站结构和导航路径），适用于信息收集、内容分析和研究任务。`
`http://mcpo:8000/github`	`github`	`创建、更新和管理GitHub仓库、文件、分支、问题和拉取请求；搜索代码、仓库、问题和用户；合并拉取请求、添加评论、获取提交历史等，覆盖了GitHub的常用操作。`
`http://mcpo:8000/edgeone-pages-mcp-server`	`HTML预览`	`部署HTML内容到EdgeOne Pages并返回公开URL，以及部署文件夹或压缩文件到EdgeOne Pages并返回公开URL。`

之后需要在右下角点击保存。

打开一个新的OpenWebUI页面，点击输入框左下角的加号图标。即可选择mcp工具了。

其他文件

OpenWebUI的gemini生图函数，由gemini-2.5-pro修改为支持imagen-3.0-generate-002。具体使用说明请看原帖：

函数代码

"""
title: Gemini Pipe
author_url: https://linux.do/u/coker/summary
author: coker
modifier_url: https://linux.do/u/zgccrui/summary
modifier: zgccrui
modifier_url: https://linux.do/u/zhang0281/summary
modifier: zhang
version: 1.1.4
license: MIT
"""

import json
import random
import httpx
import re
import base64
import io
import traceback
from typing import List, AsyncGenerator, Callable, Awaitable
from pydantic import BaseModel, Field
import time
import uuid


class Pipe:
    class Valves(BaseModel):
        GOOGLE_API_KEYS: str = Field(
            default="", description="API Keys for Google, use , to split"
        )
        BASE_URL: str = Field(
            default="https://generativelanguage.googleapis.com/v1beta",
            description="API Base Url",
        )
        OPEN_SEARCH_INFO: bool = Field(
            default=True, description="Open search info show "
        )
        GEMINI_API_MODEL: str = Field(
            default="gemini-2.0-flash-exp",
            description="API请求的模型名称，默认为 gemini-2.0-flash-exp，多模型名可使用`,`分隔",
        )
        IMAGE_HOST_URL: str = Field(
            default="https://cloudflare-imgbed-7b0.pages.dev/upload",
            description="图床URL,使用图床项目地址为 https://github.com/MarSeventh/CloudFlare-ImgBed",
        )
        IMAGE_HOST_AUTH_CODE: str = Field(default="", description="图床鉴权Key")
        RESEND_IMAGES: bool = Field(
            default=True, description="是否重新发送历史消息中的图片到大模型"
        )

    def __init__(self):
        self.type = "manifold"
        self.name = ""
        self.valves = self.Valves()
        self.OPEN_SEARCH_MODELS = ["gemini-2.0-pro-exp"]
        self.OPEN_IMAGE_OUT_MODELS = ["gemini-2.0-flash-exp"]
        self.emitter = None
        self.open_search = False
        self.open_image = False

    def pipes(self) -> List[dict]:
        models = self.valves.GEMINI_API_MODEL.split(",")
        return [
            {
                "id": model.strip(),
                "name": model.strip(),
            }
            for model in models
        ]

    def create_search_link(self, idx, web):
        return f'\n{idx:02d}: [**{web["title"]}**]({web["uri"]})'

    async def upload_image_to_host(self, image_data, mime_type):
        """上传图片到图床并返回URL"""
        try:
            file_ext = mime_type.split("/")[1] if "/" in mime_type else "jpg"
            file_like = io.BytesIO(image_data)
            # 准备多部分表单数据
            current_date = time.strftime("%Y/%m/%d")
            files = {
                "file": (
                    f"{current_date}/{uuid.uuid4().hex[:8]}.{file_ext}",
                    file_like,
                    mime_type,
                )
            }
            # 使用Valves中的图床URL和认证码
            upload_url = f"{self.valves.IMAGE_HOST_URL}?authCode={self.valves.IMAGE_HOST_AUTH_CODE}&uploadNameType=origin"
            # 上传到图床服务
            async with httpx.AsyncClient() as client:
                response = await client.post(upload_url, files=files, timeout=30)
                if response.status_code == 200:
                    # 从响应中获取路径
                    image_path = response.json()[0]["src"]
                    # 从IMAGE_HOST_URL提取基本域名
                    base_domain = "/".join(self.valves.IMAGE_HOST_URL.split("/")[:3])
                    # 构建完整URL
                    full_image_url = f"{base_domain}{image_path}"
                    return full_image_url
                else:
                    print(
                        f"图片上传失败，状态码 {response.status_code}: {response.text}"
                    )
                    return None
        except Exception as e:
            print(f"图片上传错误: {str(e)}")
            print(traceback.format_exc())
            return None

    async def download_image_from_url(self, url):
        """从URL下载图片"""
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(url, timeout=30)
                if response.status_code == 200:
                    return response.content, response.headers.get(
                        "content-type", "image/jpeg"
                    )
                else:
                    print(f"图片下载失败，状态码 {response.status_code}")
                    return None, None
        except Exception as e:
            print(f"图片下载错误: {str(e)}")
            return None, None

    async def do_parts(self, parts):
        res = ""
        if not parts or not isinstance(parts, list):
            return "Error: No parts found"
        for part in parts:
            if "text" in part:
                res += part["text"]
            if "inlineData" in part and part["inlineData"]:
                try:
                    # 解码base64图片数据
                    image_data = base64.b64decode(part["inlineData"]["data"])
                    mime_type = part["inlineData"]["mimeType"]
                    # 上传到图床
                    image_url = await self.upload_image_to_host(image_data, mime_type)
                    if image_url:
                        res += f"\n![image]({image_url})\n"
                    else:
                        res += "\n[图片上传失败]\n"
                except Exception as e:
                    print(f"图片上传错误: {str(e)}")
                    print(traceback.format_exc())
                    res += f"\n[图片上传错误: {str(e)}]\n"
        return res

    async def process_message_content(self, message):
        """处理消息内容，包括解析和重新发送图片到大模型"""
        if isinstance(message.get("content"), str):
            content = message["content"]
            # 检查内容中是否包含Markdown格式的图片链接
            img_pattern = r"!\[image\]\((.*?)\)"
            img_matches = re.findall(img_pattern, content)
            # 如果是assistant消息，包含图片链接，并启用了图片重新发送功能
            if (
                img_matches
                and message["role"] == "assistant"
                and self.valves.RESEND_IMAGES
            ):
                # 准备处理后的parts
                parts = []
                # 处理文本部分
                text_parts = re.split(img_pattern, content)
                for i, text_part in enumerate(text_parts):
                    if text_part:  # 添加非空文本
                        parts.append({"text": text_part})
                    # 如果还有对应的图片链接，尝试下载并作为图片数据发送
                    if i < len(img_matches):
                        try:
                            # 下载图片数据
                            img_data, content_type = await self.download_image_from_url(
                                img_matches[i]
                            )
                            if img_data:
                                # 获取mime类型
                                mime_type = content_type or "image/jpeg"
                                # 以base64编码添加为内联数据
                                img_base64 = base64.b64encode(img_data).decode("utf-8")
                                parts.append(
                                    {
                                        "inline_data": {
                                            "mime_type": mime_type,
                                            "data": img_base64,
                                        }
                                    }
                                )
                            else:
                                # 下载失败，保留原始文本
                                parts.append(
                                    {"text": f"\n![image]({img_matches[i]})\n"}
                                )
                        except Exception as e:
                            print(f"处理图片错误: {str(e)}")
                            # 处理失败，添加原始链接作为文本
                            parts.append({"text": f"\n![image]({img_matches[i]})\n"})
                return {
                    "role": "model" if message["role"] == "assistant" else "user",
                    "parts": parts,
                }
            else:
                # 没有图片或不需要处理，返回原始内容
                return {
                    "role": "user" if message["role"] == "user" else "model",
                    "parts": [{"text": content}],
                }
        elif isinstance(message.get("content"), list):
            parts = []
            for content in message["content"]:
                if content["type"] == "text":
                    parts.append({"text": content["text"]})
                elif content["type"] == "image_url":
                    image_url = content["image_url"]["url"]
                    if image_url.startswith("data:image"):
                        # 处理base64编码的图片
                        image_data = image_url.split(",")[1]
                        mime_type = image_url.split(";")[0].split(":")[1]
                        parts.append(
                            {
                                "inline_data": {
                                    "mime_type": mime_type,
                                    "data": image_data,
                                }
                            }
                        )
                    else:
                        # 处理外部图片URL
                        try:
                            img_data, content_type = await self.download_image_from_url(
                                image_url
                            )
                            if img_data:
                                mime_type = content_type or "image/jpeg"
                                img_base64 = base64.b64encode(img_data).decode("utf-8")
                                parts.append(
                                    {
                                        "inline_data": {
                                            "mime_type": mime_type,
                                            "data": img_base64,
                                        }
                                    }
                                )
                            else:
                                parts.append({"image_url": image_url})
                        except Exception as e:
                            print(f"处理URL图片错误: {str(e)}")
                            parts.append({"image_url": image_url})
            return {
                "role": "user" if message["role"] == "user" else "model",
                "parts": parts,
            }
        return {
            "role": "user" if message["role"] == "user" else "model",
            "parts": [{"text": "无内容"}],
        }

    async def pipe(
        self,
        body: dict,
        __event_emitter__: Callable[[dict], Awaitable[None]] = None,
    ) -> AsyncGenerator[str, None]:
        self.emitter = __event_emitter__
        self.GOOGLE_API_KEY = random.choice(
            self.valves.GOOGLE_API_KEYS.split(",")
        ).strip()
        self.base_url = self.valves.BASE_URL
        if not self.GOOGLE_API_KEY:
            yield "Error: GOOGLE_API_KEY is not set"
            return
        try:
            model_id = body["model"]
            if "." in model_id:
                model_id = model_id.split(".", 1)[1]
            messages = body["messages"]
            stream = body.get("stream", False)

            # 检查是否为imagen模型
            is_imagen_model = "imagen" in model_id

            if is_imagen_model:
                # 获取最后一条用户消息作为提示词
                user_messages = [msg for msg in messages if msg["role"] == "user"]
                if not user_messages:
                    yield "Error: 未找到用户消息作为Imagen模型的提示词"
                    return

                last_user_message = user_messages[-1]
                prompt = last_user_message["content"]

                # 处理不同格式的内容
                if isinstance(prompt, list):
                    # 如果是列表格式，提取文本部分
                    prompt_text = ""
                    for item in prompt:
                        if item.get("type") == "text":
                            prompt_text += item.get("text", "")
                    prompt = prompt_text

                # 构建imagen模型的请求
                request_data = {
                    "instances": [{"prompt": prompt}],
                    "parameters": {
                        "sampleCount": body.get("n", 1)  # 使用n参数或默认为1
                    },
                }

                # imagen模型使用predict端点
                url = f"{self.valves.BASE_URL}/models/{model_id}:predict"
                params = {"key": self.GOOGLE_API_KEY}
                headers = {"Content-Type": "application/json"}

                async with httpx.AsyncClient() as client:
                    response = await client.post(
                        url,
                        json=request_data,
                        headers=headers,
                        params=params,
                        timeout=120,
                    )

                    if response.status_code != 200:
                        yield f"Error: HTTP {response.status_code}: {response.text}"
                        return

                    data = response.json()
                    # 处理imagen响应
                    if "predictions" in data:
                        image_urls = []
                        for prediction in data["predictions"]:
                            if "bytesBase64Encoded" in prediction:
                                # 解码base64图片数据
                                image_data = base64.b64decode(
                                    prediction["bytesBase64Encoded"]
                                )
                                mime_type = "image/jpeg"  # 通常是JPEG格式
                                # 上传到图床
                                image_url = await self.upload_image_to_host(
                                    image_data, mime_type
                                )
                                if image_url:
                                    image_urls.append(f"![image]({image_url})")

                        result = (
                            "生成的图片：\n\n" + "\n\n".join(image_urls)
                            if image_urls
                            else "图像生成失败"
                        )
                        yield result
                    else:
                        yield "无法解析imagen模型的响应"
            else:
                # 原有的非imagen模型处理逻辑
                # 准备请求载荷
                contents = []
                request_data = {
                    "generationConfig": {
                        "temperature": body.get("temperature", 0.7),
                        "topP": body.get("top_p", 0.9),
                        "topK": body.get("top_k", 40),
                        "maxOutputTokens": body.get("max_tokens", 8192),
                        "stopSequences": body.get("stop", []),
                    },
                }
                for message in messages:
                    if message["role"] == "system":
                        request_data["system_instruction"] = {
                            "parts": [{"text": message["content"]}]
                        }
                    elif message["role"] != "system":
                        # 使用新的消息处理函数
                        processed_message = await self.process_message_content(message)
                        contents.append(processed_message)
                request_data["contents"] = contents
                if model_id.endswith("-search"):
                    model_id = model_id[:-7]
                    request_data["tools"] = [{"googleSearch": {}}]
                    self.open_search = True
                elif "image-generation" in model_id:
                    # model_id = model_id[:-6]
                    request_data["generationConfig"]["response_modalities"] = [
                        "Text",
                        "Image",
                    ]
                    self.open_image = True
                params = {"key": self.GOOGLE_API_KEY}
                if stream:
                    url = f"{self.valves.BASE_URL}/models/{model_id}:streamGenerateContent"
                    params["alt"] = "sse"
                else:
                    url = f"{self.valves.BASE_URL}/models/{model_id}:generateContent"
                headers = {"Content-Type": "application/json"}
                async with httpx.AsyncClient() as client:
                    if stream:
                        async with client.stream(
                            "POST",
                            url,
                            json=request_data,
                            headers=headers,
                            params=params,
                            timeout=120,
                        ) as response:
                            if response.status_code != 200:
                                yield f"Error: HTTP {response.status_code}: {(await response.aread()).decode('utf-8')}"
                                return
                            async for line in response.aiter_lines():
                                if line.startswith("data: "):
                                    try:
                                        data = json.loads(line[6:])
                                        if "candidates" in data and data["candidates"]:
                                            parts = data["candidates"][0]["content"][
                                                "parts"
                                            ]
                                            text = await self.do_parts(parts)
                                            yield text
                                            try:
                                                if (
                                                    self.open_search
                                                    and self.valves.OPEN_SEARCH_INFO
                                                    and data["candidates"][0][
                                                        "groundingMetadata"
                                                    ]["groundingChunks"]
                                                ):
                                                    yield "\n---------------------------------\n"
                                                    groundingChunks = data[
                                                        "candidates"
                                                    ][0]["groundingMetadata"][
                                                        "groundingChunks"
                                                    ]
                                                    for (
                                                        idx,
                                                        groundingChunk,
                                                    ) in enumerate(groundingChunks, 1):
                                                        if "web" in groundingChunk:
                                                            yield self.create_search_link(
                                                                idx,
                                                                groundingChunk["web"],
                                                            )
                                            except Exception as e:
                                                pass
                                    except Exception as e:
                                        # yield f"Error parsing stream: {str(e)}"
                                        pass
                    else:
                        response = await client.post(
                            url,
                            json=request_data,
                            headers=headers,
                            params=params,
                            timeout=120,
                        )
                        if response.status_code != 200:
                            yield f"Error: HTTP {response.status_code}: {response.text}"
                            return
                        data = response.json()
                        res = ""
                        if "candidates" in data and data["candidates"]:
                            parts = data["candidates"][0]["content"]["parts"]
                            res = await self.do_parts(parts)
                            try:
                                if (
                                    self.open_search
                                    and self.valves.OPEN_SEARCH_INFO
                                    and data["candidates"][0]["groundingMetadata"][
                                        "groundingChunks"
                                    ]
                                ):
                                    res += "\n---------------------------------\n"
                                    groundingChunks = data["candidates"][0][
                                        "groundingMetadata"
                                    ]["groundingChunks"]
                                    for idx, groundingChunk in enumerate(
                                        groundingChunks, 1
                                    ):
                                        if "web" in groundingChunk:
                                            res += self.create_search_link(
                                                idx, groundingChunk["web"]
                                            )
                            except Exception as e:
                                pass
                            yield res
                        else:
                            yield "No response data"
        except Exception as e:
            yield f"Error: {str(e)}\n{traceback.format_exc()}"

函数选项：

选项	值	说明
Google Api Keys	<你的Api Key>
Base Url	https://api-proxy.me/gemini/v1beta	我这里因为网络原因，使用了bbb大佬的 LLM API 代理服务
Open Search Info	Enabled
Gemini Api Model	gemini-2.0-flash-exp-image-generation,imagen-3.0-generate-002
Image Host Url	<你的图床地址>
Image Host Auth Code	<你的Auth Code>
Resend Images	Enabled

其他说明

现在我这边测试，家里云，cf tunnel+梯子反而是访问最快的方法。就很难评。
用Augment做了个小工具，可以自动设置模型列表的图标，标签和描述。导出模型列表的json放在项目根目录。之后参考readme.md就可以了。注意：这个工具完全是Vibe Coding的，而且没有经过完全测试。可能存在问题或bug。建议备份好你自己的导出数据。 zhang0281/owu_model_list_init

感谢列表

Reno大佬的mcpo教程。

bbb大佬的api代理。

应该还有其他的贴子，但是实在想不起来了…

liuzeen · 2025 年6 月 21 日 16:52

太牛了佬！看不懂先收藏回来慢慢研究

yqyan · 2025 年6 月 21 日 21:31

感谢佬友分享教程

handsome · 2025 年6 月 22 日 01:08

感谢大佬教程

konbakuyomu · 2025 年6 月 22 日 06:01

感谢大佬的无私奉献，已经收藏

话题		回复	浏览量
给openwebui的gemini加上读取网页联网和代码执行开发调优人工智能 , 软件开发	29	869	2025 年8 月 12 日
owu + imagebed 水一下？开发调优软件开发 , 作品集	6	177	2026 年1 月 9 日
openwebui gemini 绘图上传图床开发调优 Gemini , 人工智能	24	1027	2025 年5 月 8 日
Open WebUI有什么好用的函数推荐？开发调优人工智能 , 快问快答	12	1082	2025 年1 月 22 日
在Openwebui中对传入的图片进行文字识别后再交给模型处理搞七捻三人工智能	15	1145	2025 年1 月 29 日