tts

import asyncio
import logging
from typing import List

from livekit import agents, rtc
from livekit.plugins import elevenlabs, cartesia, inworld  # example plugin imports

logger = logging.getLogger("tts-fallback")
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)

class TTSProviderWrapper:
    def __init__(self, name: str, tts_instance, max_retries: int = 2, retry_delay: float = 1.0):
        self.name = name
        self.tts = tts_instance
        self.max_retries = max_retries
        self.retry_delay = retry_delay

    async def synthesize_stream(self, text: str):
        """
        Attempt to synthesize. Return async iterable/stream of audio or raise.
        Adapt this method to the actual plugin API (some return bytes, some async generators).
        """
        # Example: assume plugin provides async generator `synthesize(text)`
        attempt = 0
        last_exc = None
        while attempt <= self.max_retries:
            try:
                <http://logger.info|logger.info>(f"Trying TTS provider {self.name} (attempt {attempt+1})")
                return self.tts.synthesize(text)  # may be async generator or awaitable
            except Exception as e:
                last_exc = e
                logger.warning(f"TTS {self.name} attempt {attempt+1} failed: {e}")
                attempt += 1
                if attempt <= self.max_retries:
                    await asyncio.sleep(self.retry_delay * attempt)
        raise last_exc

async def publish_audio_stream(room: rtc.Room, audio_stream):
    """
    Publish audio stream to the LiveKit room.
    Adapt depending on audio_stream shape. Here we assume an async iterable yields audio frames/bytes.
    """
    async for chunk in audio_stream:
        await room.local_participant.publish_audio(chunk)

async def speak_with_fallback(
    room: rtc.Room,
    text: str,
    providers: List[TTSProviderWrapper],
):
    last_error = None
    for provider in providers:
        try:
            # Get an async stream/generator for audio from the provider
            audio_stream = await provider.synthesize_stream(text)
            # If synthesize_stream returned an async generator, publish it:
            await publish_audio_stream(room, audio_stream)
            <http://logger.info|logger.info>(f"Succeeded with TTS provider: {provider.name}")
            return provider.name  # success - return which provider was used
        except Exception as e:
            logger.warning(f"Provider {provider.name} failed: {e}")
            last_error = e
            # Optionally detect credit-exhaustion errors here and mark provider as disabled
            # Example:
            # if is_credit_exhaustion_error(e):
            #     disable_provider_in_registry(provider.name)
            continue
    # All providers failed
    logger.error(f"All TTS providers failed. Last error: {last_error}")
    raise last_error

# Example Agent entrypoint
async def agent_entry(ctx: agents.JobContext):
    await ctx.connect()
    room = ctx.room

    # Initialize provider wrappers in the fallback order
    providers = [
        TTSProviderWrapper("Inworld", inworld.TTS(api_key="INWORLD_KEY")),
        TTSProviderWrapper("Cartesia", cartesia.TTS(model="sonic-3", voice="v1")),
        TTSProviderWrapper("ElevenLabs", elevenlabs.TTS(api_key="ELEVEN_KEY")),
    ]

    # Example usage: speak a message, automatically falling back as needed
    try:
        used = await speak_with_fallback(room, "Hello — switching TTS providers automatically.", providers)
        <http://logger.info|logger.info>(f"Spoken using provider: {used}")
    except Exception:
        # handle global failure (log, alert, play silence, etc.)
        logger.exception("Failed to synthesize speech with all providers")

if __name__ == "__main__":
    server = agents.ServiceService()
    # register your agent entrypoint appropriately for your setup
    server.register_agent(agents.JobRequestHandler(entrypoint=agent_entry))
    server.start()

max_retries

retry_delay

chat_ctx

synthesize