diff --git a/examples/pipelines/integrations/flowise_pipeline.py b/examples/pipelines/integrations/flowise_pipeline.py
new file mode 100644
index 0000000..18f9ce1
--- /dev/null
+++ b/examples/pipelines/integrations/flowise_pipeline.py
@@ -0,0 +1,272 @@
+"""
+title: FlowiseAI Integration
+author: Claude
+author_url: https://anthropic.com
+git_url: https://github.com/open-webui/pipelines/
+description: Access FlowiseAI endpoints with customizable flows
+required_open_webui_version: 0.4.3
+requirements: requests
+version: 0.4.3
+licence: MIT
+"""
+
+from typing import List, Union, Generator, Iterator, Dict, Optional
+from pydantic import BaseModel, Field
+import requests
+import os
+import re
+import json
+from datetime import datetime
+import time
+
+from logging import getLogger
+logger = getLogger(__name__)
+logger.setLevel("DEBUG")
+
+
+class Pipeline:
+ class Valves(BaseModel):
+ API_KEY: str = Field(default="", description="FlowiseAI API key")
+ API_URL: str = Field(default="", description="FlowiseAI base URL")
+ RATE_LIMIT: int = Field(default=5, description="Rate limit for the pipeline")
+
+ FLOW_0_ENABLED: Optional[bool] = Field(default=False, description="Flow 0 Enabled (make this flow available for use)")
+ FLOW_0_ID: Optional[str] = Field(default=None, description="Flow 0 ID (the FlowiseAI flow identifier)")
+ FLOW_0_NAME: Optional[str] = Field(default=None, description="Flow 0 Name (human-readable name for the flow)")
+
+ FLOW_1_ENABLED: Optional[bool] = Field(default=False, description="Flow 1 Enabled (make this flow available for use)")
+ FLOW_1_ID: Optional[str] = Field(default=None, description="Flow 1 ID (the FlowiseAI flow identifier)")
+ FLOW_1_NAME: Optional[str] = Field(default=None, description="Flow 1 Name (human-readable name for the flow)")
+
+ FLOW_2_ENABLED: Optional[bool] = Field(default=False, description="Flow 2 Enabled (make this flow available for use)")
+ FLOW_2_ID: Optional[str] = Field(default=None, description="Flow 2 ID (the FlowiseAI flow identifier)")
+ FLOW_2_NAME: Optional[str] = Field(default=None, description="Flow 2 Name (human-readable name for the flow)")
+
+ FLOW_3_ENABLED: Optional[bool] = Field(default=False, description="Flow 3 Enabled (make this flow available for use)")
+ FLOW_3_ID: Optional[str] = Field(default=None, description="Flow 3 ID (the FlowiseAI flow identifier)")
+ FLOW_3_NAME: Optional[str] = Field(default=None, description="Flow 3 Name (human-readable name for the flow)")
+
+ FLOW_4_ENABLED: Optional[bool] = Field(default=False, description="Flow 4 Enabled (make this flow available for use)")
+ FLOW_4_ID: Optional[str] = Field(default=None, description="Flow 4 ID (the FlowiseAI flow identifier)")
+ FLOW_4_NAME: Optional[str] = Field(default=None, description="Flow 4 Name (human-readable name for the flow)")
+
+ def __init__(self):
+ self.name = "FlowiseAI Pipeline"
+
+ # Initialize valve parameters from environment variables
+ self.valves = self.Valves(
+ **{k: os.getenv(k, v.default) for k, v in self.Valves.model_fields.items()}
+ )
+
+ # Build flow mapping for faster lookup
+ self.flows = {}
+ self.update_flows()
+
+ def update_flows(self):
+ """Update the flows dictionary based on the current valve settings"""
+ self.flows = {}
+ # Iterate through each flow
+ for i in range(20): # Support up to 20 flows
+ enabled_name = f"FLOW_{i}_ENABLED"
+ if not hasattr(self.valves, enabled_name): # sequential numbering
+ break
+ enabled = getattr(self.valves, f"FLOW_{i}_ENABLED", False)
+ flow_id = getattr(self.valves, f"FLOW_{i}_ID", None)
+ flow_name = getattr(self.valves, f"FLOW_{i}_NAME", None)
+
+ if enabled and flow_id and flow_name:
+ self.flows[flow_name.lower()] = flow_id
+
+ logger.info(f"Updated flows: {list(self.flows.keys())}")
+
+ async def on_startup(self):
+ """Called when the server is started"""
+ logger.debug(f"on_startup:{self.name}")
+ self.update_flows()
+
+ async def on_shutdown(self):
+ """Called when the server is stopped"""
+ logger.debug(f"on_shutdown:{self.name}")
+
+ async def on_valves_updated(self) -> None:
+ """Called when valves are updated"""
+ logger.debug(f"on_valves_updated:{self.name}")
+ self.update_flows()
+
+ def rate_check(self, dt_start: datetime) -> bool:
+ """
+ Check time, sleep if not enough time has passed for rate
+
+ Args:
+ dt_start (datetime): Start time of the operation
+ Returns:
+ bool: True if sleep was done
+ """
+ dt_end = datetime.now()
+ time_diff = (dt_end - dt_start).total_seconds()
+ time_buffer = (1 / self.valves.RATE_LIMIT)
+ if time_diff >= time_buffer: # no need to sleep
+ return False
+ time.sleep(time_buffer - time_diff)
+ return True
+
+ def parse_user_input(self, user_message: str) -> tuple[str, str]:
+ """
+ Parse the user message to extract flow name and query
+
+ Format expected: @flow_name: query
+
+ Args:
+ user_message (str): User's input message
+
+ Returns:
+ tuple[str, str]: Flow name and query
+ """
+ # Match pattern @flow_name: query
+ pattern = r"^@([^:]+):\s*(.+)$"
+ match = re.match(pattern, user_message.strip())
+
+ if not match:
+ return None, user_message
+
+ flow_name = match.group(1).strip().lower()
+ query = match.group(2).strip()
+
+ return flow_name, query
+
+ def pipe(
+ self,
+ user_message: str,
+ model_id: str,
+ messages: List[dict],
+ body: dict
+ ) -> Union[str, Generator, Iterator]:
+ """
+ Main pipeline function. Calls a specified FlowiseAI flow with the provided query.
+
+ Format expected: @flow_name: query
+ If no flow is specified, a list of available flows will be returned.
+ """
+ logger.debug(f"pipe:{self.name}")
+
+ dt_start = datetime.now()
+ streaming = body.get("stream", False)
+ logger.warning(f"Stream: {streaming}")
+ context = ""
+
+ # Check if we have valid API configuration
+ if not self.valves.API_KEY or not self.valves.API_URL:
+ error_msg = "FlowiseAI configuration missing. Please set API_KEY and API_URL valves."
+ if streaming:
+ yield error_msg
+ else:
+ return error_msg
+
+ # Parse the user message to extract flow name and query
+ flow_name, query = self.parse_user_input(user_message)
+
+ # If no flow specified or invalid flow, list available flows
+ if not flow_name or flow_name not in self.flows:
+ available_flows = list(self.flows.keys())
+ if not available_flows:
+ no_flows_msg = "No flows configured. Enable at least one FLOW_X_ENABLED valve and set its ID and NAME."
+ if streaming:
+ yield no_flows_msg
+ else:
+ return no_flows_msg
+
+ flows_list = "\n".join([f"- @{flow}" for flow in available_flows])
+ help_msg = f"Please specify a flow using the format: @flow_name: your query\n\nAvailable flows:\n{flows_list}"
+
+ if not flow_name:
+ help_msg = "No flow specified. " + help_msg
+ else:
+ help_msg = f"Invalid flow '{flow_name}'. " + help_msg
+
+ if streaming:
+ yield help_msg
+ else:
+ return help_msg
+
+ # Get the flow ID from the map
+ flow_id = self.flows[flow_name]
+
+ if streaming:
+ yield from self.stream_retrieve(flow_id, flow_name, query, dt_start)
+ else:
+ for chunk in self.stream_retrieve(flow_id, flow_name, query, dt_start):
+ context += chunk
+ return context if context else "No response from FlowiseAI"
+
+ def stream_retrieve(
+ self, flow_id: str, flow_name: str, query: str, dt_start: datetime
+ ) -> Generator:
+ """
+ Call the FlowiseAI endpoint with the specified flow ID and query.
+
+ Args:
+ flow_id (str): The ID of the flow to call
+ flow_name (str): The name of the flow (for logging)
+ query (str): The user's query
+ dt_start (datetime): Start time for rate limiting
+
+ Returns:
+ Generator: Response chunks for streaming
+ """
+ if not query:
+ yield "Query is empty. Please provide a question or prompt for the flow."
+ return
+
+ api_url = f"{self.valves.API_URL.rstrip('/')}/api/v1/prediction/{flow_id}"
+ headers = {"Authorization": f"Bearer {self.valves.API_KEY}"}
+
+ payload = {
+ "question": query,
+ }
+
+ try:
+ logger.info(f"Calling FlowiseAI flow '{flow_name}' with query: {query}")
+
+ # Rate limiting check
+ self.rate_check(dt_start)
+
+ response = requests.post(api_url, headers=headers, json=payload)
+
+ if response.status_code != 200:
+ error_msg = f"Error from FlowiseAI: Status {response.status_code}"
+ logger.error(f"{error_msg} - {response.text}")
+ yield error_msg
+ return
+
+ try:
+ result = response.json()
+
+ # Format might vary based on flow configuration
+ # Try common response formats
+ if isinstance(result, dict):
+ if "text" in result:
+ yield result["text"]
+ elif "answer" in result:
+ yield result["answer"]
+ elif "response" in result:
+ yield result["response"]
+ elif "result" in result:
+ yield result["result"]
+ else:
+ # If no standard field found, return full JSON as string
+ yield f"```json\n{json.dumps(result, indent=2)}\n```"
+ elif isinstance(result, str):
+ yield result
+ else:
+ yield f"```json\n{json.dumps(result, indent=2)}\n```"
+
+ except json.JSONDecodeError:
+ # If not JSON, return the raw text
+ yield response.text
+
+ except Exception as e:
+ error_msg = f"Error calling FlowiseAI: {str(e)}"
+ logger.error(error_msg)
+ yield error_msg
+
+ return
\ No newline at end of file