Module livekit.plugins.speechify

Speechify plugin for LiveKit Agents

Provides Speechify text-to-speech for LiveKit voice agents. See https://docs.speechify.ai for API details.

Classes

class ChunkedStream (*,
tts: TTS,
input_text: str,
conn_options: APIConnectOptions)

Expand source code

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

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        try:
            response = await self._tts._client.audio.speech(
                **_request_kwargs(self._input_text, self._opts),
                request_options={"timeout_in_seconds": int(self._conn_options.timeout)},
            )
            output_emitter.initialize(
                request_id=utils.shortuuid(),
                sample_rate=SAMPLE_RATE,
                num_channels=NUM_CHANNELS,
                mime_type=MIME_TYPE,
            )
            timed = _timed_transcript(response.speech_marks, 0.0)
            if timed:
                output_emitter.push_timed_transcript(timed)
            output_emitter.push(base64.b64decode(response.audio_data))
            output_emitter.flush()
        except Exception as e:
            _raise_from(e)

Used by the non-streamed synthesize API, some providers support chunked http responses

Ancestors

• livekit.agents.tts.tts.ChunkedStream
• abc.ABC

class SynthesizeStream (*,
tts: TTS,
conn_options: APIConnectOptions)

Expand source code

class SynthesizeStream(tts.SynthesizeStream):
    def __init__(self, *, tts: TTS, conn_options: APIConnectOptions) -> None:
        super().__init__(tts=tts, conn_options=conn_options)
        self._tts: TTS = tts
        self._opts = replace(tts._opts)

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        request_id = utils.shortuuid()
        output_emitter.initialize(
            request_id=request_id,
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
            mime_type=MIME_TYPE,
            stream=True,
        )
        output_emitter.start_segment(segment_id=request_id)

        sent_stream = self._tts._tokenizer.stream()

        async def _forward_input() -> None:
            async for data in self._input_ch:
                if isinstance(data, self._FlushSentinel):
                    sent_stream.flush()
                    continue
                sent_stream.push_text(data)
            sent_stream.end_input()

        async def _synthesize() -> None:
            offset = 0.0
            async for ev in sent_stream:
                if not (text := ev.token.strip()):
                    continue
                self._mark_started()
                response = await self._tts._client.audio.speech(
                    **_request_kwargs(text, self._opts),
                    request_options={"timeout_in_seconds": int(self._conn_options.timeout)},
                )
                audio = base64.b64decode(response.audio_data)
                timed = _timed_transcript(response.speech_marks, offset)
                if timed:
                    output_emitter.push_timed_transcript(timed)
                output_emitter.push(audio)
                output_emitter.flush()
                offset += len(audio) / (2 * SAMPLE_RATE * NUM_CHANNELS)

            output_emitter.end_segment()

        tasks = [
            asyncio.create_task(_forward_input()),
            asyncio.create_task(_synthesize()),
        ]
        try:
            await asyncio.gather(*tasks)
        except Exception as e:
            _raise_from(e)
        finally:
            await sent_stream.aclose()
            await utils.aio.cancel_and_wait(*tasks)

Helper class that provides a standard way to create an ABC using inheritance.

Ancestors

• livekit.agents.tts.tts.SynthesizeStream
• abc.ABC

class TTS (*,
voice_id: str = 'dominic_32',
model: TTSModels = 'simba-3.2',
language: NotGivenOr[str] = NOT_GIVEN,
loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
text_normalization: NotGivenOr[bool] = NOT_GIVEN,
api_key: NotGivenOr[str] = NOT_GIVEN,
base_url: NotGivenOr[str] = NOT_GIVEN,
tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN,
client: AsyncSpeechify | None = None,
**kwargs: Any)

Expand source code

class TTS(tts.TTS):
    def __init__(
        self,
        *,
        voice_id: str = DEFAULT_VOICE_ID,
        model: TTSModels = DEFAULT_MODEL,
        language: NotGivenOr[str] = NOT_GIVEN,
        loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
        text_normalization: NotGivenOr[bool] = NOT_GIVEN,
        api_key: NotGivenOr[str] = NOT_GIVEN,
        base_url: NotGivenOr[str] = NOT_GIVEN,
        tokenizer: NotGivenOr[tokenize.SentenceTokenizer] = NOT_GIVEN,
        client: AsyncSpeechify | None = None,
        **kwargs: Any,
    ) -> None:
        """Create a new instance of Speechify TTS.

        Synthesis uses the Speechify ``/audio/speech`` endpoint, which returns
        raw PCM (24 kHz mono) together with word-level speech marks. ``stream()``
        splits input into sentences and issues one request per sentence, emitting
        audio and aligned word timestamps as each sentence completes for
        near-streaming time-to-first-audio.

        Args:
            voice_id: Id of the voice to synthesize with. The voice must support
                the chosen ``model`` (see the ``/v1/voices`` endpoint). Defaults
                to ``dominic_32``.
            model: Synthesis model. One of ``simba-english``,
                ``simba-multilingual``, ``simba-3.0`` or ``simba-3.2``. Defaults
                to ``simba-3.2``.
            language: BCP-47 language code of the input (e.g. ``en-US``).
            loudness_normalization: Normalize output loudness to a standard
                level. Increases latency slightly when enabled.
            text_normalization: Expand numbers, dates, etc. into words before
                synthesis. Increases latency slightly when enabled.
            api_key: Speechify API key. Falls back to the ``SPEECHIFY_API_KEY``
                environment variable.
            base_url: Override the Speechify API base URL.
            tokenizer: Sentence tokenizer used to chunk input in ``stream()``.
            client: A preconfigured ``AsyncSpeechify`` client. When provided,
                ``api_key`` and ``base_url`` are ignored.
            **kwargs: Catches deprecated parameters. A warning is logged for
                any recognised deprecated name.
        """
        super().__init__(
            capabilities=tts.TTSCapabilities(streaming=True, aligned_transcript=True),
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
        )

        self._owns_client = client is None
        if client is not None:
            self._client = client
        else:
            resolved_key = api_key if is_given(api_key) else os.environ.get("SPEECHIFY_API_KEY")
            if not resolved_key:
                raise ValueError(
                    "Speechify API key is required, either as the api_key argument "
                    "or via the SPEECHIFY_API_KEY environment variable"
                )
            # Fixed httpx.AsyncClient default header so every request the SDK
            # issues is attributed to this integration, regardless of call site.
            # Timeout/limits mirror the openai plugin's owned-client defaults —
            # httpx's own 5s default is too short for longer synthesis requests.
            self._httpx_client = httpx.AsyncClient(
                headers={CALLER_HEADER: "livekit"},
                timeout=httpx.Timeout(connect=15.0, read=30.0, write=30.0, pool=5.0),
                limits=httpx.Limits(
                    max_connections=50, max_keepalive_connections=50, keepalive_expiry=120
                ),
            )
            self._client = AsyncSpeechify(
                token=resolved_key,
                base_url=base_url if is_given(base_url) else None,
                httpx_client=self._httpx_client,
            )

        self._tokenizer = tokenizer if is_given(tokenizer) else tokenize.basic.SentenceTokenizer()
        self._opts = _TTSOptions(
            voice_id=voice_id,
            model=model,
            language=language,
            loudness_normalization=loudness_normalization,
            text_normalization=text_normalization,
        )

        _check_deprecated_args(kwargs)

    @property
    def model(self) -> str:
        return self._opts.model if is_given(self._opts.model) else "unknown"

    @property
    def provider(self) -> str:
        return "Speechify"

    async def aclose(self) -> None:
        if self._owns_client:
            await self._httpx_client.aclose()

    async def list_voices(self) -> list[Voice]:
        """List the voices available for the configured Speechify account."""
        sdk_voices: list[GetVoice] = await self._client.voices.list()
        return [_voice_from_sdk(v) for v in sdk_voices]

    def update_options(
        self,
        *,
        voice_id: NotGivenOr[str] = NOT_GIVEN,
        model: NotGivenOr[TTSModels] = NOT_GIVEN,
        language: NotGivenOr[str] = NOT_GIVEN,
        loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
        text_normalization: NotGivenOr[bool] = NOT_GIVEN,
    ) -> None:
        if is_given(voice_id):
            self._opts.voice_id = voice_id
        if is_given(model):
            self._opts.model = model
        if is_given(language):
            self._opts.language = language
        if is_given(loudness_normalization):
            self._opts.loudness_normalization = loudness_normalization
        if is_given(text_normalization):
            self._opts.text_normalization = text_normalization

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

    def stream(
        self,
        *,
        conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
    ) -> SynthesizeStream:
        return SynthesizeStream(tts=self, conn_options=conn_options)

Helper class that provides a standard way to create an ABC using inheritance.

Create a new instance of Speechify TTS.

Synthesis uses the Speechify /audio/speech endpoint, which returns raw PCM (24 kHz mono) together with word-level speech marks. stream() splits input into sentences and issues one request per sentence, emitting audio and aligned word timestamps as each sentence completes for near-streaming time-to-first-audio.

Args

voice_id

Id of the voice to synthesize with. The voice must support the chosen model (see the /v1/voices endpoint). Defaults to dominic_32.

model

Synthesis model. One of simba-english, simba-multilingual, simba-3.0 or simba-3.2. Defaults to simba-3.2.

language

BCP-47 language code of the input (e.g. en-US).

loudness_normalization

Normalize output loudness to a standard level. Increases latency slightly when enabled.

text_normalization

Expand numbers, dates, etc. into words before synthesis. Increases latency slightly when enabled.

api_key

Speechify API key. Falls back to the SPEECHIFY_API_KEY environment variable.

base_url

Override the Speechify API base URL.

tokenizer

Sentence tokenizer used to chunk input in stream().

client

A preconfigured AsyncSpeechify client. When provided, api_key and base_url are ignored.

**kwargs

Catches deprecated parameters. A warning is logged for any recognised deprecated name.

Ancestors

• livekit.agents.tts.tts.TTS
• abc.ABC
• EventEmitter
• typing.Generic

Instance variables

prop model : str

Expand source code

@property
def model(self) -> str:
    return self._opts.model if is_given(self._opts.model) else "unknown"

Get the model name/identifier for this TTS instance.

Returns

The model name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their model information.

prop provider : str

Expand source code

@property
def provider(self) -> str:
    return "Speechify"

Get the provider name/identifier for this TTS instance.

Returns

The provider name if available, "unknown" otherwise.

Note

Plugins should override this property to provide their provider information.

Methods

async def aclose(self) ‑> None

Expand source code

async def aclose(self) -> None:
    if self._owns_client:
        await self._httpx_client.aclose()

async def list_voices(self) ‑> list[livekit.plugins.speechify.tts.Voice]

Expand source code

async def list_voices(self) -> list[Voice]:
    """List the voices available for the configured Speechify account."""
    sdk_voices: list[GetVoice] = await self._client.voices.list()
    return [_voice_from_sdk(v) for v in sdk_voices]

List the voices available for the configured Speechify account.

def stream(self,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.speechify.tts.SynthesizeStream

Expand source code

def stream(
    self,
    *,
    conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
) -> SynthesizeStream:
    return SynthesizeStream(tts=self, conn_options=conn_options)

def synthesize(self,
text: str,
*,
conn_options: APIConnectOptions = APIConnectOptions(max_retry=3, retry_interval=2.0, timeout=10.0)) ‑> livekit.plugins.speechify.tts.ChunkedStream

Expand source code

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

def update_options(self,
*,
voice_id: NotGivenOr[str] = NOT_GIVEN,
model: NotGivenOr[TTSModels] = NOT_GIVEN,
language: NotGivenOr[str] = NOT_GIVEN,
loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
text_normalization: NotGivenOr[bool] = NOT_GIVEN) ‑> None

Expand source code

def update_options(
    self,
    *,
    voice_id: NotGivenOr[str] = NOT_GIVEN,
    model: NotGivenOr[TTSModels] = NOT_GIVEN,
    language: NotGivenOr[str] = NOT_GIVEN,
    loudness_normalization: NotGivenOr[bool] = NOT_GIVEN,
    text_normalization: NotGivenOr[bool] = NOT_GIVEN,
) -> None:
    if is_given(voice_id):
        self._opts.voice_id = voice_id
    if is_given(model):
        self._opts.model = model
    if is_given(language):
        self._opts.language = language
    if is_given(loudness_normalization):
        self._opts.loudness_normalization = loudness_normalization
    if is_given(text_normalization):
        self._opts.text_normalization = text_normalization

Inherited members

• EventEmitter:
  • emit
  • off
  • on
  • once

class Voice (id: str,
type: VoiceType,
display_name: str,
gender: Gender,
avatar_image: str | None,
models: list[TTSModels],
locale: str)

Expand source code

@dataclass
class Voice:
    id: str
    type: VoiceType
    display_name: str
    gender: Gender
    avatar_image: str | None
    models: list[TTSModels]
    locale: str

Voice(id: 'str', type: 'VoiceType', display_name: 'str', gender: 'Gender', avatar_image: 'str | None', models: 'list[TTSModels]', locale: 'str')

Instance variables

var avatar_image : str | None

var display_name : str

var gender : Literal['male', 'female', 'neutral']

var id : str

var locale : str

var models : list[typing.Literal['simba-english', 'simba-multilingual', 'simba-3.0', 'simba-3.2']]

var type : Literal['shared', 'personal']