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:
        output_emitter.initialize(
            request_id=utils.shortuuid(),
            sample_rate=SAMPLE_RATE,
            num_channels=NUM_CHANNELS,
            mime_type=MIME_TYPE,
        )
        try:
            await _synthesize_text(
                self._tts._client,
                self._input_text,
                self._opts,
                self._conn_options,
                output_emitter,
                offset=0.0,
            )
            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()
                sentence_bytes = await _synthesize_text(
                    self._tts._client,
                    text,
                    self._opts,
                    self._conn_options,
                    output_emitter,
                    offset=offset,
                )
                output_emitter.flush()
                offset += sentence_bytes / (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.

        Both ``synthesize()`` and ``stream()`` use the
        ``/v1/audio/stream/with-timestamps`` endpoint, which streams raw PCM
        (24 kHz mono) together with word-level speech marks, so audio and aligned
        timestamps are emitted as they arrive. ``stream()`` splits its input into
        sentences and issues one streaming request per sentence. Speech marks are
        produced by the streaming-native models (``simba-3.0`` and ``simba-3.2``);
        the legacy ``simba-english`` and ``simba-multilingual`` models do not
        serve the streaming route and use one non-streamed ``/v1/audio/speech``
        request per input (``synthesize()``) or per sentence (``stream()``).

        Requests share one keep-alive connection, which ``prewarm()`` opens
        before the first request.

        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``. Voices whose id carries a ``_32`` suffix are
                part of the curated simba-3.2 roster; a warning is logged when
                they are paired with any other model.
            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
        self._prewarm_task: asyncio.Task[None] | None = None
        self._base_url = base_url if is_given(base_url) else SpeechifyEnvironment.DEFAULT.value
        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 httpx2.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 —
            # the client's own 5s default is too short for longer synthesis requests.
            # speechify-api types this argument as httpx.AsyncClient and only
            # retries httpx.ConnectError. request/stream still work against
            # httpx2; _raise_from maps the httpx2 errors onto APIError so the
            # TTS retry loop runs.
            self._httpx_client = httpx.AsyncClient(
                headers={CALLER_HEADER: "livekit", CALLER_VERSION_HEADER: __version__},
                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=self._base_url,
                httpx_client=self._httpx_client,  # type: ignore[arg-type]
            )

        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)
        _warn_voice_model_compat(self._opts.voice_id, self._opts.model)

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

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

    def prewarm(self) -> None:
        """Open the HTTPS connection to the Speechify API ahead of the first request.

        Sends an unauthenticated ``GET /health`` through the connection pool the
        synthesis requests use, so the first reply of a session does not pay for
        DNS, TCP and TLS setup. Does nothing when a preconfigured ``client`` was
        passed in, because the plugin does not own that client's connections.
        """
        if not self._owns_client:
            return

        async def _prewarm() -> None:
            try:
                await self._httpx_client.get(self._base_url.rstrip("/") + PREWARM_PATH)
            except Exception:
                pass

        # A prewarm still in flight keeps its task, so aclose() can cancel it.
        if self._prewarm_task is None or self._prewarm_task.done():
            self._prewarm_task = asyncio.create_task(_prewarm())

    async def aclose(self) -> None:
        if self._prewarm_task is not None:
            await utils.aio.cancel_and_wait(self._prewarm_task)
        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
        _warn_voice_model_compat(self._opts.voice_id, self._opts.model)

    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.

Both synthesize() and stream() use the /v1/audio/stream/with-timestamps endpoint, which streams raw PCM (24 kHz mono) together with word-level speech marks, so audio and aligned timestamps are emitted as they arrive. stream() splits its input into sentences and issues one streaming request per sentence. Speech marks are produced by the streaming-native models (simba-3.0 and simba-3.2); the legacy simba-english and simba-multilingual models do not serve the streaming route and use one non-streamed /v1/audio/speech request per input (synthesize()) or per sentence (stream()).

Requests share one keep-alive connection, which prewarm() opens before the first request.

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. Voices whose id carries a _32 suffix are part of the curated simba-3.2 roster; a warning is logged when they are paired with any other model.

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._prewarm_task is not None:
        await utils.aio.cancel_and_wait(self._prewarm_task)
    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 prewarm(self) ‑> None

Expand source code

def prewarm(self) -> None:
    """Open the HTTPS connection to the Speechify API ahead of the first request.

    Sends an unauthenticated ``GET /health`` through the connection pool the
    synthesis requests use, so the first reply of a session does not pay for
    DNS, TCP and TLS setup. Does nothing when a preconfigured ``client`` was
    passed in, because the plugin does not own that client's connections.
    """
    if not self._owns_client:
        return

    async def _prewarm() -> None:
        try:
            await self._httpx_client.get(self._base_url.rstrip("/") + PREWARM_PATH)
        except Exception:
            pass

    # A prewarm still in flight keeps its task, so aclose() can cancel it.
    if self._prewarm_task is None or self._prewarm_task.done():
        self._prewarm_task = asyncio.create_task(_prewarm())

Open the HTTPS connection to the Speechify API ahead of the first request.

Sends an unauthenticated GET /health through the connection pool the synthesis requests use, so the first reply of a session does not pay for DNS, TCP and TLS setup. Does nothing when a preconfigured client was passed in, because the plugin does not own that client's connections.

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
    _warn_voice_model_compat(self._opts.voice_id, self._opts.model)

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']