> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/open-webui/open-webui/llms.txt
> Use this file to discover all available pages before exploring further.
# Pipelines
> Extend Open WebUI with the Pipelines plugin framework
Pipelines provide a powerful plugin framework that allows you to seamlessly integrate custom logic and Python libraries into Open WebUI. The framework acts as middleware between the UI and your language models, enabling you to modify requests and responses.
## Overview
Pipelines can be used to:
* Add custom logic before and after model inference
* Implement rate limiting and usage monitoring
* Filter and transform messages
* Integrate third-party services
* Enable live translation
* Monitor and log conversations
Pipelines run as a separate service and communicate with Open WebUI through OpenAI-compatible API endpoints.
## Installation
Clone and set up the Pipelines repository:
```bash theme={null}
git clone https://github.com/open-webui/pipelines.git
cd pipelines
pip install -r requirements.txt
```
Launch the Pipelines service:
```bash theme={null}
python main.py
```
By default, the server runs on `http://localhost:9099`
In Open WebUI admin settings, add the Pipelines URL as an OpenAI API endpoint:
* Navigate to **Admin Panel → Settings → Connections**
* Add new OpenAI API URL: `http://localhost:9099`
* Set an API key (any string will work for local development)
## Pipeline Architecture
Pipelines implement two main filter types:
### Inlet Filters
Process requests **before** they reach the model:
```python theme={null}
class Pipeline:
async def inlet(self, body: dict, user: dict) -> dict:
# Modify the request body
messages = body.get("messages", [])
# Add system context
messages.insert(0, {
"role": "system",
"content": "You are a helpful assistant."
})
body["messages"] = messages
return body
```
### Outlet Filters
Process responses **after** the model generates them:
```python theme={null}
class Pipeline:
async def outlet(self, body: dict, user: dict) -> dict:
# Modify the response
messages = body.get("messages", [])
# Log or transform response
if messages:
last_message = messages[-1]
print(f"Model response: {last_message}")
return body
```
## Managing Pipelines
### Upload Pipeline
Admin users can upload custom pipeline files through the API:
```python theme={null}
import requests
url = "http://localhost:8080/api/pipelines/upload"
files = {"file": open("my_pipeline.py", "rb")}
data = {"urlIdx": 0}
response = requests.post(url, files=files, data=data)
```
### Add Pipeline from URL
Install pipelines directly from a URL:
```python theme={null}
import requests
url = "http://localhost:8080/api/pipelines/add"
payload = {
"url": "https://github.com/user/repo/pipeline.py",
"urlIdx": 0
}
response = requests.post(url, json=payload)
```
### List Available Pipelines
Retrieve all configured pipeline instances:
```python theme={null}
import requests
response = requests.get("http://localhost:8080/api/pipelines/list")
pipelines = response.json()
```
## Valves Configuration
Pipelines support configurable parameters called "valves":
```python theme={null}
from pydantic import BaseModel, Field
class Pipeline:
class Valves(BaseModel):
priority: int = Field(
default=0,
description="Pipeline priority (higher runs first)"
)
enabled: bool = Field(
default=True,
description="Enable or disable this pipeline"
)
api_key: str = Field(
default="",
description="External API key"
)
def __init__(self):
self.valves = self.Valves()
```
### Get Pipeline Valves
```python theme={null}
import requests
response = requests.get(
"http://localhost:8080/api/pipelines/my-pipeline/valves",
params={"urlIdx": 0}
)
valves = response.json()
```
### Update Pipeline Valves
```python theme={null}
import requests
url = "http://localhost:8080/api/pipelines/my-pipeline/valves/update"
payload = {
"priority": 10,
"enabled": True,
"api_key": "sk-..."
}
response = requests.post(url, json=payload, params={"urlIdx": 0})
```
## Common Use Cases
```python theme={null}
import time
from collections import defaultdict
class Pipeline:
def __init__(self):
self.user_requests = defaultdict(list)
self.max_requests = 10
self.time_window = 60 # seconds
async def inlet(self, body: dict, user: dict) -> dict:
user_id = user["id"]
current_time = time.time()
# Clean old requests
self.user_requests[user_id] = [
req_time for req_time in self.user_requests[user_id]
if current_time - req_time < self.time_window
]
# Check rate limit
if len(self.user_requests[user_id]) >= self.max_requests:
raise Exception("Rate limit exceeded. Please try again later.")
self.user_requests[user_id].append(current_time)
return body
```
```python theme={null}
import re
class Pipeline:
def __init__(self):
self.blocked_patterns = [
r"\b(password|secret|api[_-]?key)\b",
r"\b\d{3}-\d{2}-\d{4}\b", # SSN pattern
]
async def inlet(self, body: dict, user: dict) -> dict:
messages = body.get("messages", [])
for message in messages:
content = message.get("content", "")
for pattern in self.blocked_patterns:
if re.search(pattern, content, re.IGNORECASE):
raise Exception("Message contains sensitive information")
return body
```
```python theme={null}
import logging
from datetime import datetime
class Pipeline:
def __init__(self):
self.logger = logging.getLogger(__name__)
async def inlet(self, body: dict, user: dict) -> dict:
self.logger.info(f"Request from {user['email']} at {datetime.now()}")
return body
async def outlet(self, body: dict, user: dict) -> dict:
messages = body.get("messages", [])
if messages:
response_length = len(str(messages[-1]))
self.logger.info(f"Response to {user['email']}: {response_length} chars")
return body
```
## Pipeline Priority
Multiple pipelines can be chained together. The execution order is determined by priority:
```python theme={null}
class Pipeline:
class Valves(BaseModel):
priority: int = Field(default=0)
def __init__(self):
self.valves = self.Valves(priority=10) # Higher runs first
```
**Inlet filters** execute in descending priority order (highest first).
**Outlet filters** execute in ascending priority order (lowest first).
## Troubleshooting
Ensure your pipeline file:
* Is a valid Python file (`.py` extension)
* Contains a `Pipeline` class
* Implements `inlet()` and/or `outlet()` methods
* Handles exceptions appropriately
### Common Issues
**Pipeline not appearing in Open WebUI:**
* Check that the Pipelines server is running
* Verify the OpenAI API URL is correctly configured
* Ensure the pipeline file was uploaded successfully
**Pipeline errors:**
* Check server logs for Python exceptions
* Validate that all dependencies are installed
* Ensure valve configuration is valid
## Next Steps
* Explore the [Pipelines Examples](https://github.com/open-webui/pipelines/tree/main/examples)
* Learn about [Functions](/advanced/functions) for Python function calling
* Configure [Tools](/advanced/tools) for extended capabilities