# Open WebUI
[Open WebUI](https://github.com/open-webui/open-webui) is an extensible, feature-rich, and user-friendly *self-hosted AI platform* designed to operate entirely offline.
[Functions](https://docs.openwebui.com/features/plugin/functions/) are like plugins for Open WebUI. We can create a custom [Pipe Function](https://docs.openwebui.com/features/plugin/functions/pipe) that process inputs and generate responses by invoking Flowise Prediction API before returning results to the user. Through this, Flowise can be used in Open WebUI.
## Setup
1. First, have Open WebUI up and running, you can refer to the [Quickstart](https://docs.openwebui.com/getting-started/quick-start/) guide. From the left bottom, click your profile and **Admin Panel**
2. Open **Functions** tab, and add a new Function.
3. Name the Function, and add the following code:
```python
"""
title: Flowise Integration for OpenWebUI
Requirements:
- Flowise API URL (set via FLOWISE_API_URL)
- Flowise API Key (set via FLOWISE_API_KEY)
"""
from pydantic import BaseModel, Field
from typing import Optional, Dict, Any, List, Union, Generator, Iterator
import requests
import json
import os
class Pipe:
class Valves(BaseModel):
flowise_url: str = Field(
default=os.getenv("FLOWISE_API_URL", ""),
description="Flowise URL",
)
flowise_api_key: str = Field(
default=os.getenv("FLOWISE_API_KEY", ""),
description="Flowise API key for authentication",
)
def __init__(self):
self.type = "manifold"
self.id = "flowise_chat"
self.valves = self.Valves()
# Validate required settings
if not self.valves.flowise_url:
print(
"⚠️ Please set your Flowise URL using the FLOWISE_API_URL environment variable"
)
if not self.valves.flowise_api_key:
print(
"⚠️ Please set your Flowise API key using the FLOWISE_API_KEY environment variable"
)
def pipes(self):
if self.valves.flowise_api_key and self.valves.flowise_url:
try:
headers = {
"Authorization": f"Bearer {self.valves.flowise_api_key}",
"Content-Type": "application/json",
}
r = requests.get(
f"{self.valves.flowise_url}/api/v1/chatflows?type=AGENTFLOW",
headers=headers,
)
models = r.json()
return [
{
"id": model["id"],
"name": model["name"],
}
for model in models
]
except Exception as e:
return [
{
"id": "error",
"name": str(e),
},
]
else:
return [
{
"id": "error",
"name": "API Key not provided.",
},
]
def _process_message_content(self, message: dict) -> str:
"""Process message content, handling text for now"""
if isinstance(message.get("content"), list):
processed_content = []
for item in message["content"]:
if item["type"] == "text":
processed_content.append(item["text"])
return " ".join(processed_content)
return message.get("content", "")
def pipe(
self, body: dict, __user__: Optional[dict] = None, __metadata__: dict = None
):
try:
stream_enabled = body.get("stream", True)
session_id = (__metadata__ or {}).get("chat_id") or "owui-session"
# model can be "flowise." or just ""
model_name = body.get("model", "")
dot = model_name.find(".")
model_id = model_name[dot + 1 :] if dot != -1 else model_name
messages = body.get("messages") or []
if not messages:
raise Exception("No messages found in request body")
question = self._process_message_content(messages[-1])
data = {
"question": question,
"overrideConfig": {"sessionId": session_id},
"streaming": stream_enabled,
}
headers = {
"Authorization": f"Bearer {self.valves.flowise_api_key}",
"Content-Type": "application/json",
"Accept": "text/event-stream" if stream_enabled else "application/json",
}
url = f"{self.valves.flowise_url}/api/v1/prediction/{model_id}"
with requests.post(
url, json=data, headers=headers, stream=stream_enabled, timeout=60
) as r:
r.raise_for_status()
if stream_enabled:
# Ensure correct decoding for SSE (prevents ’ etc.)
r.encoding = "utf-8"
for raw_line in r.iter_lines(decode_unicode=True):
if not raw_line:
continue
line = raw_line.strip()
# Skip keep-alives or non-data fields
if not line.startswith("data:"):
continue
payload = line[5:].strip()
if payload in ("[DONE]", '"[DONE]"'):
break
# Flowise usually sends {"event":"token","data":"..."}
try:
obj = json.loads(payload)
except json.JSONDecodeError:
# Occasionally plain text arrives—stream it anyway
if payload:
yield payload
continue
if isinstance(obj, dict):
if obj.get("event") == "token":
token = obj.get("data") or ""
if token:
yield token
else:
# Some versions send {"data":{"text":"..."}}
data_field = obj.get("data")
if isinstance(data_field, dict):
text = data_field.get("text")
if text:
yield text
return # end streaming
# Non-streaming fallback
resp = r.json()
return (
resp.get("text") or (resp.get("data") or {}).get("text", "") or ""
)
except requests.HTTPError as http_err:
try:
detail = http_err.response.text[:500]
except Exception:
detail = ""
return f"HTTP error from Flowise: {http_err.response.status_code} {detail}"
except Exception as e:
return f"Error in Flowise pipe: {e}"
```
4. After Function has been saved, enable it, and click the settings button to put in your Flowise URL and Flowise API Key:
5. Now when you refresh and click New Chat, you will be able to see the list of flows. You can modify the code to show:
* Only Agentflows V2: `f"{self.valves.flowise_url}/api/v1/chatflows?type=AGENTFLOW"`
* Only Chatflows: `f"{self.valves.flowise_url}/api/v1/chatflows?type=CHATFLOW"`
* Only Assistants: `f"{self.valves.flowise_url}/api/v1/chatflows?type=ASSISTANT"`
6. Test:
---
# Agent Instructions: Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter:
```
GET https://docs.flowiseai.com/integrations/3rd-party-platform-integration/open-webui.md?ask=
```
The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.