Pipecat integration

Native Shunyalabs STT and TTS services for Pipecat pipelines. Drop ShunyalabsSTTService and ShunyalabsTTSService into any Pipecat pipeline and get real-time streaming ASR with 46 speakers across 23 languages - no glue code required.

Installation

Install the package from PyPI:

pip install pipecat-shunyalabsai

To include a transport (e.g. Daily WebRTC):

pip install pipecat-shunyalabsai pipecat-ai[daily]

Authentication

You need a Shunya Labs API key. Create one in the console - sign in at console.shunyalabs.ai, open API Keys, and click Create New Key (it is shown only once). Provide it as an environment variable (recommended) or pass it directly to the service classes:

export SHUNYALABS_API_KEY="your-api-key"

stt = ShunyalabsSTTService(api_key="your-api-key")
tts = ShunyalabsTTSService(api_key="your-api-key")

Quick start

A minimal pipeline wiring Shunyalabs STT → OpenAI LLM → Shunyalabs TTS on a local audio transport:

import asyncio, os
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.local.audio import LocalAudioTransport
from pipecat_shunyalabs import ShunyalabsSTTService, ShunyalabsTTSService

async def main():
    transport = LocalAudioTransport()

    stt = ShunyalabsSTTService(
        api_key=os.environ["SHUNYALABS_API_KEY"],
        language="en",
    )

    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o",
    )

    tts = ShunyalabsTTSService(
        api_key=os.environ["SHUNYALABS_API_KEY"],
        voice="Rajesh",
        language="en",
        style="<Conversational>",
    )

    pipeline = Pipeline([transport.input(), stt, llm, tts, transport.output()])
    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
    await PipelineRunner().run(task)

if __name__ == "__main__":
    asyncio.run(main())

STT - ShunyalabsSTTService

Real-time streaming speech-to-text over a persistent WebSocket connection. Supports 23 Indian and international languages with optional automatic language detection.

Parameters

Frame mapping

from pipecat_shunyalabs import ShunyalabsSTTService

stt = ShunyalabsSTTService(
    language="hi",       # Hindi; use "auto" for detection
    sample_rate=16000,
)

TTS - ShunyalabsTTSService

Streaming text-to-speech over WebSocket. The plugin maintains a persistent WebSocket session that is reused across synthesis requests, streaming audio chunks back as TTSAudioRawFrame frames. Supports 46 speakers across 23 languages - any speaker can synthesise in any language.

Parameters

Style tags

Text formatting

The service automatically prepends the style tag before sending to the API:

tts = ShunyalabsTTSService(voice="Rajesh", style="<Happy>")
# Input:  "Welcome!"
# Sent:   "<Happy> Welcome!"

Turn-taking

The gateway does its own endpointing, and can tell you when a speaker has finished rather than leaving you to infer it from a second voice-activity detector reading the same audio. Set emit_turn_frames=True to surface that as Pipecat turn frames.

from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair, LLMUserAggregatorParams,
)
from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies
from pipecat_shunyalabs import ShunyalabsSTTService

stt = ShunyalabsSTTService(
    language="en",
    emit_turn_frames=True,          # both halves are required
)

aggregator = LLMContextAggregatorPair(
    context,
    user_params=LLMUserAggregatorParams(
        user_turn_strategies=ExternalUserTurnStrategies(),
    ),
)

Two caveats worth knowing before you rely on this:

• The turn-start frame is derived from the first interim result, so it is only as prompt as decode_every_ms (640 ms by default). That is fine for deciding whose turn it is, but too slow to drive barge-in - keep a transport voice-activity detector for that.
• A speaker who never pauses produces no turn-end signal at all, because the gateway cuts the segment at its maximum length instead of endpointing. If your agent must respond regardless, pair this with your own timeout.

Latency tuning

Measured against the production services. Figures are first-audio or time-to-final on a warm connection; treat them as the shape of the trade-off rather than a promise for your network.

The one setting that matters most

endpoint_silence_ms is pure waiting: it is wall-clock delay after the speaker stops, before a transcript can be finalised. It defaults to 700 ms. A deployment running it at 3000 measured just over three seconds of perceived turn latency from this setting alone.

Lower is not simply better. At 250-300 ms a speaker who pauses mid-sentence - "a table for... four" - is cut off, and the agent answers half a sentence. Raise it if that happens; lower it if turns feel sluggish.

Taking the decision yourself

For full control, leave endpoint_silence_ms high as a backstop and call commit() when your own logic decides the speaker is done - a voice-activity detector, a push-to-talk release, a semantic turn detector. The gateway finalises immediately and keeps the connection open.

stt = ShunyalabsSTTService(
    language="en",
    endpoint_silence_ms=1500,   # backstop only
)

# ... when your own detector says the speaker finished:
await stt.commit()              # finalises now, connection stays open

Measured: 231 ms from commit() to the finalised transcript. Note this replaces the connection teardown that would otherwise happen per turn - a fresh handshake to the gateway measured around 300 ms from India, against an ASR that finalises in 39 ms.

Synthesis: what actually helps

Measured on a 130-character reply, warm connection, each setting isolated:

Two results worth reading carefully. min_buffer_frames does little on its own, because the buffer is a threshold in audio duration and the service usually delivers faster than real time - 480 ms of audio can arrive in a few tens of milliseconds. And clause_first on its own is slower: a short opening clause does not reach the default 12-frame threshold until the next piece arrives, which is why the two settings belong together.

So quality carries most of the saving, and it is the one with a quality cost. Use it for throwaway acknowledgements, not for content a caller has to hear correctly.

Barge-in

When the caller interrupts, the service discards the in-flight utterance and keeps the session open, rather than dropping and re-establishing the connection. Measured at 88 ms to discard, with the next reply starting 246 ms later on the already-warm session. Pipecat triggers this automatically on an interruption; call it directly if you detect the interruption first:

await tts.discard_current_turn()

Telephony

For 8 kHz phone audio, pass sample_rate=8000 to the STT service. The gateway accepts it natively, so there is no resampling stage on the way in - measured at 251 ms to final, against 231 ms for the same speech at 16 kHz. Lower min_send_bytes to 1600 to keep the send buffer at 100 ms.

Full pipeline example

A complete voice agent with Shunyalabs STT and TTS, OpenAI LLM, and the Daily WebRTC transport:

import asyncio, os
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.openai_llm_context import (
    OpenAILLMContext, OpenAILLMContextAggregator,
)
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.services.daily import DailyParams, DailyTransport
from pipecat_shunyalabs import ShunyalabsSTTService, ShunyalabsTTSService

async def run_voice_agent(room_url: str, token: str):
    transport = DailyTransport(
        room_url, token, "Shunyalabs Agent",
        DailyParams(audio_out_enabled=True, transcription_enabled=False),
    )

    stt = ShunyalabsSTTService(
        api_key=os.environ["SHUNYALABS_API_KEY"],
        language="auto",
        sample_rate=16000,
    )

    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        model="gpt-4o",
    )

    messages = [{"role": "system", "content": "You are a helpful voice assistant."}]
    context = OpenAILLMContext(messages)
    context_aggregator = llm.create_context_aggregator(context)

    tts = ShunyalabsTTSService(
        api_key=os.environ["SHUNYALABS_API_KEY"],
        voice="Rajesh",
        language="hi",
        style="<Conversational>",
    )

    pipeline = Pipeline([
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ])

    task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True))

    @transport.event_handler("on_first_participant_joined")
    async def on_first_participant_joined(transport, participant):
        await task.queue_frames([context_aggregator.user().get_context_frame()])

    await PipelineRunner().run(task)

if __name__ == "__main__":
    asyncio.run(run_voice_agent(
        room_url=os.environ["DAILY_ROOM_URL"],
        token=os.environ["DAILY_TOKEN"],
    ))

Multilingual example

# Hindi conversational bot
tts = ShunyalabsTTSService(voice="Rajesh", language="hi", style="<Conversational>")

# English news-style bot
tts = ShunyalabsTTSService(voice="Varun", language="en", style="<News>")

Custom endpoints

Both services can be repointed without changing code or upgrading the package. Resolution precedence: explicit argument → endpoint returned by the token service → environment variable → built-in default.

export SHUNYALABS_ASR_WS_URL="wss://<host>/v1/realtime"
export SHUNYALABS_TTS_WS_URL="wss://<host>/v1/realtime"

stt = ShunyalabsSTTService(url="wss://<host>/v1/realtime")
tts = ShunyalabsTTSService(url="wss://<host>/v1/realtime")

Error reference

Troubleshooting