In conversational AI, flexibility and control are essential. Sometimes out-of-the-box text-to-speech (TTS) plugins aren’t enough—perhaps you need to deploy a new open TTS model, support a specific language, or host your own voice for privacy or branding. LiveKit’s agent framework makes it possible to use your own TTS backend by implementing a custom plugin. This post walks through our working example and explains—in detail—how you can do the same.

Why Build a Custom TTS Plugin?

While LiveKit offers integrations for many popular TTS vendors, real-world needs often demand:

• Hosting your own models (e.g., for privacy, control, or experimenting with new open-source research).
• Supporting new voices, emotion controls, or languages unavailable in commercial APIs.
• Prototyping and rapid iteration on novel voice tech.

What You Need ?

• A running TTS API server (any REST endpoint that takes text and returns audio is eligible).
• The ability to process audio as raw bytes (e.g., PCM-encoded).

The Core: TTS Plugin Structure in LiveKit

All TTS plugins in LiveKit are composed of two main parts:

1. The TTS Plugin Class

This subclass tells LiveKit how to call your TTS service.

Key methods & arguments:

• voice and denoise (custom options for your service).
• base_url for the API endpoint.
• sample_rate, num_channels matching your model’s output.
• synthesize()—returns a custom ChunkedStream that streams synthesized audio.

from livekit.agents import tts, utils, APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS
from dataclasses import dataclass
import httpx

@dataclass
class _TTSOptions:
    voice: str
    denoise: bool

class TTS(tts.TTS):
    def __init__(self, *, 
                 voice: str = "my_voice.wav",
                 denoise: bool = False, 
                 base_url: str):
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=False),
            sample_rate=24000,    # Must match your model/server output!
            num_channels=1,
        )
        self._client = httpx.AsyncClient(timeout=60.0)
        self._base_url = base_url
        self._opts = _TTSOptions(voice=voice, denoise=denoise)

    def synthesize(self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS):
        # Returns a ChunkedStream to LiveKit
        return ChunkedStream(
            tts=self,
            input_text=text,
            conn_options=conn_options,
        )

2. The ChunkedStream Class

Handles sending the text to your server and streaming audio bytes to LiveKit’s audio emitter as soon as they arrive.

Workflow:

• LiveKit calls your plugin’s synthesize()
• That launches the _run() function on ChunkedStream
• Audio bytes are streamed chunk-by-chunk to the agent pipeline

class ChunkedStream(tts.ChunkedStream):
    def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions):
        super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
        self._tts: TTS = tts

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        output_emitter.initialize(
            request_id=utils.shortuuid(),
            sample_rate=self._tts.sample_rate,
            num_channels=self._tts.num_channels,
            mime_type="audio/raw",  # Raw PCM bytes
        )
        payload = {
            "text": self.input_text,
            "voice": self._tts._opts.voice,
        }
        # Make streaming POST request to your API
        async with self._tts._client.stream(
            "POST",
            f"{self._tts._base_url}/v1/audio/speech",
            json=payload,
        ) as response:
            response.raise_for_status()
            async for chunk in response.aiter_bytes():
                output_emitter.push(chunk)
        # End stream only after all audio is sent
        output_emitter.flush()

FastAPI Example: Streaming a TTS Model

Your API must:

• Accept a JSON body with at least a text field (optionally voice, etc.)
• Yield raw PCM-encoded audio bytes (audio/raw) in the response.

Here’s the core pattern:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

class SynthesisRequest(BaseModel):
    text: str
    voice: str

@app.post("/v1/audio/speech")
async def stream_tts_audio(request: SynthesisRequest):
    ... # Prepare your model/voice/etc.
    async def audio_streamer():
        ... # Logic to run your model and yield PCM16 frames
        while True:
            chunk = ... # Next chunk of PCM16
            if chunk is None:
                break
            yield chunk
    return StreamingResponse(audio_streamer(), media_type="audio/raw")

Important:

• PCM audio: Usually 16-bit (int16), mono, and sample rate should match your agent plugin.
• You can use any model backend (e.g., your finetuned TTS, open-source like Kyutai, Coqui, etc.)

How to Integrate

In your agent setup (e.g., agent.py), use your TTS plugin like this:

from tts import TTS

class MyAssistant(Agent):
    def __init__(self):
        custom_tts = TTS(base_url="http://<your-server>:8000")
        super().__init__(
            ...,
            tts=custom_tts,
            ...
        )

Key Benefits

• Total model freedom: Use any TTS model you want—even experimental or regional ones.
• Immediate updates: Swap voices/models by updating your server—no plugin code changes required.
• Data/Privacy control: Host and control all voice output in-house, if needed.
• Customization: Support fine-tuning for language, dialect, accent, or unique emotions/expressiveness.

Summary Table

<table>

With this approach, you can build powerful and flexible voice agents—fully tailored to your product, research, or language needs. The only requirements: your server must stream audio in real time and match the expected format and endpoint pattern.