Integrations
SonexTTSService extends Pipecat's official TTSService base class, so it works exactly like CartesiaTTSService or ElevenLabsTTSService — drop it into any Pipecat transport (SmallWebRTC, FastAPI WebSocket, Twilio, Exotel, Vobiz); no Daily required.
pip install pipecat-ai pipecat-sonex
Requires Python 3.10+. Tested with pipecat-ai v1.7.0.
from pipecat_sonex import SonexTTSService
tts = SonexTTSService(
api_key="vsk_...", # or set SONEX_API_KEY env var
voice="9b8fsavyez", # from GET /v1/voices — required, no default
)
# Drop into any pipeline, same as any other pipecat TTS service
pipeline = Pipeline([
transport.input(),
stt,
context_aggregator.user(),
llm,
tts, # ← SonexTTSService here
transport.output(),
context_aggregator.assistant(),
])api_keystrRequiredSonexLabs API key (vsk_...). Falls back to SONEX_API_KEY env var.
voicestrRequiredVoice ID from GET /v1/voices. No default.
languagestrOptionalLeave unset — Pāṇini TTS auto-detects language from the input text.
speedfloatOptionalSpeaking rate multiplier. Practical range 0.75–1.5. Default 1.0.
sample_rateintOptionalPipeline output rate in Hz. Not sent to the API — pipecat resamples to this value. Default 24000. Use 8000 for telephony.
endpointstrOptionalAPI base URL. Default https://api.sonexlabs.com.
settingsSonexTTSSettingsOptionalRuntime-updatable settings, takes precedence over constructor args.
The processor buffers LLMTextFrame tokens until a sentence boundary, then:
{ "input": "...", "voice": "...", "response_format": "wav" } to the /v1/speech/stream endpointTTSAudioRawFrame frames downstream to the transport at the configured sample rateA pooled aiohttp.ClientSession (via TCPConnector keep-alive) reuses connections across requests instead of reconnecting per sentence. It also handles TTSSpeakFrame for direct synthesis, strips markdown formatting, and skips non-speakable tokens.
import os, asyncio
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.services.groq.llm import GroqLLMService
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport
from pipecat_sonex import SonexTTSService
async def run_bot(webrtc_connection, stt):
transport = SmallWebRTCTransport(
webrtc_connection,
params=TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(),
),
)
llm = GroqLLMService(
api_key=os.getenv("GROQ_API_KEY"),
model="llama-3.3-70b-versatile",
)
tts = SonexTTSService(
api_key=os.getenv("SONEX_API_KEY"),
voice=os.getenv("SONEX_VOICE_ID"),
speed=1.0,
sample_rate=16000,
)
pipeline = Pipeline([
transport.input(),
stt,
llm,
tts,
transport.output(),
])
runner = PipelineRunner()
await runner.run(PipelineTask(
pipeline,
params=PipelineParams(allow_interruptions=True),
))Set sample_rate=8000 to match the telephony μ-law transport output rate. Pipecat's SOXR resampler handles 24 kHz → 8 kHz downsampling transparently. VobizFrameSerializer ships bundled in pipecat_sonex.vobiz — no extra package needed.
tts = SonexTTSService(
api_key=os.getenv("SONEX_API_KEY"),
voice="9b8fsavyez",
sample_rate=8000,
)SONEX_API_KEY=vsk_your_key_here SONEX_VOICE_ID=your_voice_id