""" title: Runware.ai FLUX Manifold description: Integrate Runware.ai's FLUX and custom models into Open-WebUI. Supports LoRAs and custom models. author: dotJustin author_url: https://github.com/dot-Justin funding_url: https://github.com/open-webui version: 0.1.0 license: MIT requirements: pydantic, requests environment_variables: RUNWARE_API_KEY """ """ WARNING: Owui makes 2 extra API calls per conversation (tags and title) by default, counting as usage without useful output. To avoid this, create a single chat and return to it for generating images. - Alternatively, you can go to Settings > Interface and turn off both "Title Auto-Generation" and "Chat Tags Auto-Generation." HOW TO USE LoRA'S (Only Flux.1 Dev and Schnell, sorry.): Select the base model for the LoRA. In the prompt, the first word should be the LoRA id. EXAMPLE PROMPT: - No LoRA | minecraft steve sitting down - With LoRA | civitai:654379@756115 minecraft steve sitting down HOW TO ADD CUSTOM MODELS: 1. Go to https://my.runware.ai/models to find available models 2. In your environment settings, locate the CUSTOM_MODELS field 3. Add models using the format: Name1|model_id1,Name2|model_id2 EXAMPLES: - Single model: "Realistic|runware:123@1" - Multiple models: "Realistic|runware:123@1,Anime|runware:456@1" """ import os from typing import Iterator, List, Union from open_webui.utils.misc import get_last_user_message from pydantic import BaseModel, Field import json import requests import uuid class Pipe: class Valves(BaseModel): RUNWARE_API_KEY: str = Field(default="", description="Your Runware.ai API key") CUSTOM_MODELS: str = Field( default="", description="Find models at https://my.runware.ai/models. Add additional models in this format: Name1|model_id1,Name2|model_id2", ) HEIGHT: int = Field( default=512, title="Image Height", description="Image height (must be divisible by 64, range: 512-2048)", ) WIDTH: int = Field( default=512, title="Image Width", description="Image width (must be divisible by 64, range: 512-2048)", ) STEPS: int = Field( default=20, description="Number of denoising steps (range: 1-100)" ) CFG_SCALE: float = Field( default=7.0, description="Guidance scale (range: 0-30). Higher values = closer to prompt", ) SEED: int = Field( default=-1, description="Random seed (-1 for random, or specify for reproducible results)", ) NUMBER_RESULTS: int = Field( default=1, description="Number of images to generate per request (range: 1-4)", ) INCLUDE_COST: bool = Field( default=True, description="Show generation cost in response" ) class UserValves(BaseModel): NEGATIVE_PROMPT: str = Field( default="", description="Negative prompt to guide what to avoid in generation", ) USE_PROMPT_WEIGHTING: bool = Field( default=False, description="Enable prompt weighting syntax (e.g., word+, (word:1.2))", ) def __init__(self): self.type = "manifold" self.id = "runware_ai" self.name = "Runware FLUX" self.valves = self.Valves(RUNWARE_API_KEY=os.getenv("RUNWARE_API_KEY", "")) self.user_valves = self.UserValves() self.api_endpoint = "https://api.runware.ai/v1" def get_runware_models(self) -> List[dict]: """Get list of Runware models including FLUX and custom models""" models = [ {"id": "runware:100@1", "name": "Flux.1 Schnell"}, {"id": "runware:101@1", "name": "Flux.1 Dev"}, ] if self.valves.CUSTOM_MODELS: try: for model_entry in self.valves.CUSTOM_MODELS.split(","): if "|" in model_entry: name, model_id = model_entry.split("|", 1) if model_id.strip() not in [m["id"] for m in models]: models.append( {"id": model_id.strip(), "name": name.strip()} ) except Exception as e: print(f"Error parsing custom models: {e}") return models def pipes(self) -> List[dict]: return self.get_runware_models() async def pipe( self, body: dict, __event_emitter__=None ) -> Union[str, Iterator[str]]: try: if not self.valves.RUNWARE_API_KEY: await __event_emitter__( { "type": "status", "data": { "description": "Error: RUNWARE_API_KEY is not set", "done": True, }, } ) return "Error: RUNWARE_API_KEY is not set" positive_prompt = get_last_user_message(body["messages"]) selected_model = body.get("model", "runware:100@1") await __event_emitter__( { "type": "status", "data": { "description": "Setting up generation request...", "done": False, }, } ) # Check if message starts with a CivitAI ID for LoRA lora_config = None if positive_prompt.startswith("civitai:"): parts = positive_prompt.split(None, 1) # Split on first whitespace if len(parts) == 2: lora_id = parts[0] positive_prompt = parts[1] # Rest of the prompt lora_config = [{"model": lora_id, "weight": 1.0}] await __event_emitter__( { "type": "status", "data": { "description": f"Adding LoRA {lora_id} to model {selected_model}...", "done": False, }, } ) payload = [ {"taskType": "authentication", "apiKey": self.valves.RUNWARE_API_KEY}, { "taskType": "imageInference", "taskUUID": str(uuid.uuid4()), "outputType": "URL", "outputFormat": "JPG", "positivePrompt": positive_prompt, "height": self.valves.HEIGHT, "width": self.valves.WIDTH, "model": selected_model, "steps": self.valves.STEPS, "CFGScale": self.valves.CFG_SCALE, "numberResults": self.valves.NUMBER_RESULTS, "includeCost": self.valves.INCLUDE_COST, }, ] # Add LoRA configuration if specified if lora_config: payload[1]["lora"] = lora_config # Debug message for LoRA configuration await __event_emitter__( { "type": "status", "data": { "description": f"Using LoRA configuration: {json.dumps(lora_config)}", "done": False, }, } ) if self.valves.SEED != -1: payload[1]["seed"] = self.valves.SEED if self.user_valves.NEGATIVE_PROMPT: payload[1]["negativePrompt"] = self.user_valves.NEGATIVE_PROMPT if self.user_valves.USE_PROMPT_WEIGHTING: payload[1]["usePromptWeighting"] = True await __event_emitter__( { "type": "status", "data": { "description": "Sending request to Runware...", "done": False, }, } ) # Debug message to show the complete payload await __event_emitter__( { "type": "status", "data": { "description": f"Request payload: {json.dumps(payload[1], indent=2)}", "done": False, }, } ) response = requests.post( self.api_endpoint, json=payload, headers={ "Content-Type": "application/json", "Accept": "application/json", }, ) response.raise_for_status() data = response.json() if "error" in data: error_msg = data.get("errorMessage", "Unknown error occurred") await __event_emitter__( { "type": "status", "data": { "description": f"Error: {error_msg}", "done": True, }, } ) return f"API Error: {error_msg}" await __event_emitter__( { "type": "status", "data": { "description": "Processing response...", "done": False, }, } ) message = "" if "data" in data: for result in data["data"]: if result.get("taskType") == "imageInference": if "imageURL" in result: message += f"\n" if ( self.valves.INCLUDE_COST and "cost" in result and __event_emitter__ ): await __event_emitter__( { "type": "status", "data": { "description": f"Generation cost: ${result['cost']:.4f}", "done": True, }, } ) elif "error" in result: message += f"Error: {result['error']}\n" elif ( result.get("taskType") == "authentication" and "error" in result ): return f"Authentication Error: {result['error']}" return message or "No images were generated in the response" except Exception as e: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Error: {str(e)}", "done": True, }, } ) return f"Error: {str(e)}"