A custom TTS for LiveKit Agents: give your voice agent any voice

How a TTS plugin works

A plugin subclasses tts.TTS and returns a ChunkedStream from synthesize. The stream’s _run method sends the request and pushes the audio bytes to an AudioEmitter, which decodes Ogg, mp3 or wav itself and cuts them into frames. We checked this against LiveKit Agents 1.8: when a TTS does not stream text in, the session wraps it in a StreamAdapter that splits the model’s answer into sentences, so each sentence becomes one short request and the first one plays while the model is still writing.

pip install livekit-agents aiohttp

The plugin

Save this as superclonica_tts.py next to your agent:

import os
import aiohttp
from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, APIConnectOptions, APIStatusError, tts, utils

URL = "https://superclonica.com/api/v1/speech"


class SuperClonicaTTS(tts.TTS):
    def __init__(self, voice: str, language: str = "en"):
        super().__init__(capabilities=tts.TTSCapabilities(streaming=False), sample_rate=44100, num_channels=1)
        self.voice, self.language = voice, language
        self._session = None

    def synthesize(self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS):
        return SuperClonicaStream(tts=self, input_text=text, conn_options=conn_options)

    def session(self):
        if self._session is None:
            key = os.environ["SUPERCLONICA_KEY"]
            self._session = aiohttp.ClientSession(headers={"Authorization": "Bearer " + key})
        return self._session

    async def aclose(self):
        if self._session:
            await self._session.close()


class SuperClonicaStream(tts.ChunkedStream):
    async def _run(self, output_emitter: tts.AudioEmitter):
        engine = self._tts
        body = {"text": self.input_text, "voice": engine.voice, "language": engine.language, "stream": True}
        async with engine.session().post(URL, json=body) as answer:
            if answer.status != 200:
                raise APIStatusError(await answer.text(), status_code=answer.status)
            output_emitter.initialize(
                request_id=utils.shortuuid(), sample_rate=44100, num_channels=1, mime_type="audio/ogg"
            )
            async for chunk in answer.content.iter_chunked(4096):
                output_emitter.push(chunk)
            output_emitter.flush()

The request asks for "stream": true, and the API answers with Ogg that arrives while it is being voiced. The emitter decodes it and resamples to the 44.1 kHz the class declares. Any other answer raises APIStatusError: the framework retries 429 and server errors according to its connection options and gives up at once on errors such as a wrong key.

Use it in a session

Pass the class as tts when you build the AgentSession, with the speech recognition and language model plugins you already use:

import os
from livekit.agents import AgentSession
from superclonica_tts import SuperClonicaTTS

session = AgentSession(
    stt=...,  # your speech recognition plugin
    llm=...,  # your language model plugin
    tts=SuperClonicaTTS(voice=os.environ["VOICE_ID"], language="en"),
)

Set SUPERCLONICA_KEY to a key from the speech API page and VOICE_ID to the ID shown on the page of the voice you chose. Everything else in your agent stays the same.

Limits to know

The API speaks plain HTTPS with chunked audio, not a WebSocket, and returns no word timestamps, so aligned transcripts are off. It speaks 31 languages but not Chinese yet. A plan gives seconds of speech in every 10-minute window for all calls together: the free key a minute a day for development, Starter at $5 a month up to 3 minutes per window, Pro 20 minutes, enough for several busy calls at once.