""" id: gemini_tts_read_aloud title: Gemini TTS Read Aloud author: Mohammad Sadeghian author_url: https://github.com/MSC72m funding_url: https://github.com/open-webui version: 0.1.0 license: MIT icon_url: data:image/svg+xml;base64,PHN2ZyB3aWR0aD0iMzIiIGhlaWdodD0iMzIiIHZpZXdCb3g9IjAgMCAzMiAzMiIgZmlsbD0ibm9uZSIgeG1sbnM9Imh0dHA6Ly93d3cudzMub3JnLzIwMDAvc3ZnIj4KICA8Y2lyY2xlIGN4PSIxNiIgY3k9IjE2IiByPSIxNSIgZmlsbD0iI0ZGQjUwMCIgc3Ryb2tlPSIjRkY5MDAwIiBzdHJva2Utd2lkdGg9IjIiLz4KICA8cGF0aCBkPSJNOSAxMXY5aDNsNiA2VjVsLTYgNkg5eiIgZmlsbD0iI0ZGRkZGRiIvPgogIDxwYXRoIGQ9Im0yMCAxMmMxLjUgMS41IDEuNSA0LjUgMCA2IiBzdHJva2U9IiNGRkZGRkYiIHN0cm9rZS13aWR0aD0iMi41IiBzdHJva2UtbGluZWNhcD0icm91bmQiLz4KICA8cGF0aCBkPSJtMjIuNSA5LjVjMi41IDIuNSAyLjUgNy41IDAgMTAiIHN0cm9rZT0iI0ZGRkZGRiIgc3Ryb2tlLXdpZHRoPSIyLjUiIHN0cm9rZS1saW5lY2FwPSJyb3VuZCIvPgo8L3N2Zz4K required_open_webui_version: 0.3.10 """ import requests import base64 from pydantic import BaseModel, Field from typing import Callable, Union, Any, Dict DEBUG = False class Action: class Valves(BaseModel): GEMINI_API_KEY: str = Field( default="", description="Your Gemini API key from Google AI Studio (starts with 'AIza...'). Get it from: https://aistudio.google.com/app/apikey", ) GEMINI_TTS_MODEL: str = Field( default="gemini-2.5-flash-preview-tts", description="Gemini TTS model to use. Options: gemini-2.5-flash-preview-tts (faster) or gemini-2.5-pro-preview-tts (higher quality).", ) VOICE_NAME: str = Field( default="Kore", description="Default voice name. Available: Kore, Aoede, Charon, Fenrir, Puck.", ) def __init__(self): self.valves = self.Valves() def status_object( self, description: str = "Unknown State", status: str = "in_progress", done: bool = False, ) -> Dict: return { "type": "status", "data": { "status": status, "description": description, "done": done, }, } async def action( self, body: dict, __user__: dict = {}, __event_emitter__: Callable[[dict], Any] = None, __event_call__: Callable[[dict], Any] = None, ) -> None: if DEBUG: print(f"Debug: Gemini TTS action invoked") try: if __event_emitter__: await __event_emitter__( self.status_object("Initializing Gemini Text-to-Speech") ) if not self.valves.GEMINI_API_KEY: raise ValueError("Gemini API key is not set") if not self.valves.GEMINI_API_KEY.startswith("AIza"): raise ValueError("Invalid Gemini API key format") if "id" not in __user__: raise ValueError("User not authenticated") # Use the pre-configured voice from settings instead of asking user selected_voice_name = self.valves.VOICE_NAME selected_voice_id = self.valves.VOICE_NAME if DEBUG: print(f"Debug: Using pre-configured voice: {selected_voice_name}") messages = body.get("messages", []) assistant_message = next( ( message.get("content") for message in reversed(messages) if message.get("role") == "assistant" and message.get("content", "").strip() ), None, ) if not assistant_message: raise ValueError("No assistant message to convert") # Remove the language check since Gemini TTS actually supports multiple languages # The test script shows Persian works fine if __event_emitter__: await __event_emitter__( self.status_object("Generating speech with Gemini TTS") ) # Make the TTS API call url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.valves.GEMINI_TTS_MODEL}:generateContent?key={self.valves.GEMINI_API_KEY}" payload = { "contents": [ {"parts": [{"text": assistant_message[:1000]}]} # Limit text length ], "generationConfig": { "response_modalities": ["AUDIO"], "speech_config": { "voice_config": { "prebuilt_voice_config": {"voice_name": selected_voice_id} } }, }, } headers = { "Content-Type": "application/json", } if DEBUG: print(f"Debug: Making request to: {url}") print(f"Debug: Using model: {self.valves.GEMINI_TTS_MODEL}") print(f"Debug: Using voice: {selected_voice_id}") response = requests.post(url, json=payload, headers=headers, timeout=30) response.raise_for_status() if response.status_code == 200: response_data = response.json() if DEBUG: print(f"Debug: Response keys: {list(response_data.keys())}") print(f"Debug: Full response structure: {response_data}") # Extract audio data from Gemini response candidates = response_data.get("candidates", []) if not candidates: if DEBUG: print("Debug: No candidates found in response") raise ValueError("No audio generated by the API") if DEBUG: print(f"Debug: Found {len(candidates)} candidates") candidate = candidates[0] content = candidate.get("content", {}) parts = content.get("parts", []) if DEBUG: print(f"Debug: Candidate structure: {candidate}") print(f"Debug: Content structure: {content}") print(f"Debug: Parts structure: {parts}") print(f"Debug: Number of parts: {len(parts)}") audio_data = None mime_type_found = None # Extract audio data - try multiple approaches for i, part in enumerate(parts): if DEBUG: print(f"Debug: Processing part {i}: {part}") if "inlineData" in part: inline_data = part["inlineData"] mime_type = inline_data.get("mimeType", "") if DEBUG: print(f"Debug: Found inlineData with mimeType: {mime_type}") print(f"Debug: InlineData keys: {list(inline_data.keys())}") # Check for any audio MIME type if "audio" in mime_type.lower(): audio_content = inline_data.get("data") if audio_content: try: audio_data = base64.b64decode(audio_content) mime_type_found = mime_type if DEBUG: print( f"Debug: Successfully decoded audio data, size: {len(audio_data)} bytes" ) print(f"Debug: MIME type: {mime_type}") break except Exception as decode_error: if DEBUG: print( f"Debug: Failed to decode audio data: {decode_error}" ) continue # Also check for direct audio field (alternative format) if "audio" in part: if DEBUG: print(f"Debug: Found direct audio field in part") audio_content = part["audio"] if isinstance(audio_content, str): try: audio_data = base64.b64decode(audio_content) mime_type_found = "audio/wav" # Default assumption if DEBUG: print( f"Debug: Successfully decoded direct audio data, size: {len(audio_data)} bytes" ) break except Exception as decode_error: if DEBUG: print( f"Debug: Failed to decode direct audio data: {decode_error}" ) continue if not audio_data: if DEBUG: print("Debug: No audio data found in any part") print(f"Debug: Full response for debugging: {response_data}") raise ValueError("No valid audio content found in the response") # Use OpenWebUI's file system for audio playback if __event_emitter__: await __event_emitter__( self.status_object( "Generated successfully", status="complete", done=True ) ) # Create a file object for the audio data import tempfile import os # Create temporary file with tempfile.NamedTemporaryFile( delete=False, suffix=".wav" ) as temp_file: temp_file.write(audio_data) temp_file_path = temp_file.name # Send as file attachment await __event_emitter__( { "type": "files", "data": { "files": [ { "name": "gemini_tts_audio.wav", "type": mime_type_found or "audio/wav", "size": len(audio_data), "data": temp_file_path, } ] }, } ) # Clean up temp file os.unlink(temp_file_path) else: raise ValueError(f"Unexpected API response: {response.text}") except Exception as e: if DEBUG: print(f"Debug: Error in action method: {str(e)}") if __event_emitter__: await __event_emitter__( self.status_object(f"Error: {str(e)}", status="error", done=True) )