A custom TTS service for Pipecat: give your voice bot any voice

How a Pipecat TTS service works

A TTS service subclasses TTSService and implements run_tts, an async generator that receives a piece of text and yields audio frames of 16-bit PCM. We checked this against Pipecat 1.14: by default the service collects the model’s text into sentences, so run_tts is called once per sentence and the first one plays while the model is still writing. The base class also has a helper that strips a WAV header and resamples to the pipeline’s rate.

pip install pipecat-ai aiohttp

The service

Save this as superclonica_pipecat.py next to your bot:

import os
import aiohttp
from pipecat.frames.frames import ErrorFrame
from pipecat.services.settings import TTSSettings
from pipecat.services.tts_service import TTSService

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


class SuperClonicaTTSService(TTSService):
    def __init__(self, *, session: aiohttp.ClientSession, voice: str, language: str = "en", **kwargs):
        super().__init__(
            push_start_frame=True,
            push_stop_frames=True,
            settings=TTSSettings(model=None, voice=voice, language=language),
            **kwargs,
        )
        self._session = session
        self._voice, self._language = voice, language

    def can_generate_metrics(self):
        return True

    async def run_tts(self, text: str, context_id: str):
        headers = {"Authorization": "Bearer " + os.environ["SUPERCLONICA_KEY"]}
        body = {"text": text, "voice": self._voice, "language": self._language, "format": "wav"}
        async with self._session.post(URL, json=body, headers=headers) as answer:
            if answer.status != 200:
                yield ErrorFrame(error=f"SuperClonica: {answer.status} {await answer.text()}")
                return
            await self.start_tts_usage_metrics(text)
            async for frame in self._stream_audio_frames_from_iterator(
                answer.content.iter_chunked(self.chunk_size), strip_wav_header=True, context_id=context_id
            ):
                await self.stop_ttfb_metrics()
                yield frame

The service asks the API for WAV, 16-bit mono PCM at 44.1 kHz behind a standard 44-byte header, and passes the bytes to the helper, which drops the header, resamples to the output rate of your pipeline, such as 16 or 24 kHz, and yields the frames. If the API answers with an error, for example 429 when a plan window is used up, the service yields an ErrorFrame with the status and message.

Use it in your bot

Create the service with an aiohttp session and the ID shown on the page of the voice you chose, and put it in your Pipeline in place of the TTS service you use now:

import os
import aiohttp
from superclonica_pipecat import SuperClonicaTTSService

session = aiohttp.ClientSession()
tts = SuperClonicaTTSService(session=session, voice=os.environ["VOICE_ID"], language="en")
# put tts in your Pipeline where your current TTS service is

Set SUPERCLONICA_KEY to a key from the speech API page. The transport, speech recognition and language model of your bot stay the same.

Latency and plans

For a short sentence the WAV arrives in one piece when it is voiced; for texts longer than about half a minute of speech the API streams it. In our measurements speech starts from about 300 ms plus network time. 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. Supertonic 3 speaks 31 languages but not Chinese yet, and the API returns no word timestamps.