Pipecat Integration

KugelAudio provides an official TTS service for Pipecat, enabling high-quality voice synthesis in your voice AI pipelines.

Python only. Pipecat’s pipeline and TTSService framework is Python-only; there is no JavaScript/TypeScript equivalent to subclass, so the KugelAudio JS SDK does not ship a Pipecat service. Pipecat’s JS package (@pipecat-ai/client-js) is a browser client that connects to a Python Pipecat server over WebRTC/RTVI: run the KugelAudio Pipecat service (below) on that Python server and connect your JS/TS front-end to it. For a fully server-side JS/TS voice agent, use the LiveKit integration, which the JS SDK supports natively via kugelaudio/livekit.

Why Use KugelAudio with Pipecat?

Native service: Drop-in TTSService for Pipecat pipelines
Persistent WebSocket: Connection reuse keeps the handshake off the hot path
Built-in metrics: Automatic TTFB (Pipecat’s name for time-to-first-audio) and usage metrics
Low latency: streaming TTS built for real-time agents. See Latency for how to measure TTFA on your deployment.

Installation

pip install "kugelaudio[pipecat]"

This installs the KugelAudio SDK with its Pipecat dependency (pipecat-ai>=0.0.62 on Python 3.11 or newer). The integration supports both Pipecat 0.x and 1.x.

Quick Start

A complete local voice bot for Pipecat 1.x: microphone in, Deepgram STT, OpenAI LLM, KugelAudio TTS, speaker out. Install the Pipecat services it uses:

pip install "kugelaudio[pipecat]" "pipecat-ai[deepgram,local,openai,silero]"

Save this as bot.py:

import asyncio
import os

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
)
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.local.audio import (
    LocalAudioTransport,
    LocalAudioTransportParams,
)

from kugelaudio.pipecat import KugelAudioTTSService


async def main():
    transport = LocalAudioTransport(
        LocalAudioTransportParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            audio_out_sample_rate=24000,
            vad_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        )
    )
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(
            model="gpt-4o-mini",
            system_instruction="You are a helpful voice assistant. Keep answers short.",
        ),
    )
    tts = KugelAudioTTSService(
        model="kugel-3",
        voice_id=1071,
        sample_rate=24000,  # must match the transport's output rate
        language="en",      # normalization language (default: the voice's)
    )
    tts.prewarm()  # needs the running event loop, so call it inside main()

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

    task = PipelineTask(pipeline)
    await task.queue_frames([TTSSpeakFrame("Hi! How can I help you?")])
    await PipelineRunner().run(task)


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

Set KUGELAUDIO_API_KEY, DEEPGRAM_API_KEY and OPENAI_API_KEY, then run python bot.py. Use headphones: the local audio transport has no echo cancellation, so speaker output can feed back into the microphone. For Pipecat 0.x, or another transport, STT or LLM, keep the tts lines and build the rest with the APIs of your Pipecat version.

Configuration

Service Parameters

Parameter	Type	Default	Description
`api_key`	`str`	`KUGELAUDIO_API_KEY` env	Your KugelAudio API key
`model`	`str`	`kugel-3`	TTS model (`kugel-3`), see Models
`voice_id`	`int`	required	Voice ID to use for synthesis
`sample_rate`	`int`	`24000`	Output sample rate in Hz
`cfg_scale`	`float`	`2.0`	CFG scale, clamped to `[1.2, 2.5]`
`max_new_tokens`	`int`	`2048`	Maximum tokens to generate (`1` to `2048`)
`language`	`str \| None`	`None`	ISO 639-1 language code (e.g., `en`, `de`). Unset: the voice’s primary language
`normalize`	`bool`	`True`	Apply text normalization
`project_id`	`int \| None`	`None`	Project whose pronunciation dictionaries apply. Unset: no dictionary applies
`dictionary_ids`	`list[int] \| None`	`None`	Dictionaries by ID; `[]` disables them. A non-empty list requires `project_id`
`region`	`str \| None`	`None`	Set `"eu"` for the direct EU endpoint; overridden by `base_url`
`base_url`	`str`	`https://api.kugelaudio.com`	API base URL
`on_word_timestamps`	`Callable[[str, list[WordTimestamp]], None] \| None`	`None`	Callback receiving the context ID and each batch of word timestamps. Setting it turns word timestamps on, see Word timestamps

Supported Sample Rates

Rate	Notes
`24000`	Native rate (recommended)
`22050`	CD quality
`16000`	Wideband telephony
`8000`	Narrowband telephony

Lower rates are resampled server-side. Any other value raises ValueError in the constructor.

Performance

Set language when the text is not in the voice’s primary language, and call prewarm() inside your async setup code (it needs a running event loop and does nothing otherwise) so the first reply does not pay the WebSocket handshake. The service then reuses that connection across run_tts() calls and reconnects transparently if it drops. On Pipecat 1.x each assistant turn still opens a new server-side context; the service creates it when the LLM starts responding, so the setup overlaps the LLM’s time-to-first-token. The KugelAudio TTFA: log line measures text send to first audio chunk on the WebSocket, without LLM or STT time. See Latency for how to measure end to end and Chunking & per-segment latency for how the server chunks text.

Usage Patterns

Updating the Voice at Runtime

# Switch voice mid-conversation (closes the cached WebSocket first)
await tts.set_voice("300")

Pipeline Frame Flow

The KugelAudioTTSService emits standard Pipecat frames, in this order:

TTSStartedFrame: audio generation has begun
TTSAudioRawFrame: raw PCM audio chunks (16-bit, mono), one or more
TTSStoppedFrame: audio generation is complete
ErrorFrame: if an error occurs during synthesis

Metrics Support

The service reports Pipecat’s TTFB metric (Pipecat’s name for time-to-first-audio, measured from request to first audio chunk) and TTS usage (characters per request) automatically; tts.can_generate_metrics() returns True.

Removing Background Noise

KugelAudioEnhanceFilter puts KugelAudio speech enhancement (Clarity, clarity-1) on the transport’s audio input, the audio_in_filter slot Pipecat uses for noise filters. It removes background noise before VAD and STT hear the user. Pass speaker= with a clean 2 to 8 second sample of one person and it keeps only that voice. The Quick Start bot with enhancement on the microphone input. Save it as bot.py, set the same three API keys, and run python bot.py:

import asyncio
import os

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import TTSSpeakFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
)
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.local.audio import (
    LocalAudioTransport,
    LocalAudioTransportParams,
)

from kugelaudio import load_audio
from kugelaudio.pipecat import KugelAudioEnhanceFilter, KugelAudioTTSService


async def main():
    transport = LocalAudioTransport(
        LocalAudioTransportParams(
            audio_in_enabled=True,
            audio_in_filter=KugelAudioEnhanceFilter(),  # reads KUGELAUDIO_API_KEY
            # keep one voice: KugelAudioEnhanceFilter(speaker=load_audio("caller.wav"))
            audio_out_enabled=True,
            audio_out_sample_rate=24000,
            vad_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
        )
    )
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
    llm = OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(
            model="gpt-4o-mini",
            system_instruction="You are a helpful voice assistant. Keep answers short.",
        ),
    )
    tts = KugelAudioTTSService(
        model="kugel-3", voice_id=1071, sample_rate=24000, language="en"
    )
    tts.prewarm()

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

    task = PipelineTask(pipeline)
    await task.queue_frames([TTSSpeakFrame("Hi! How can I help you?")])
    await PipelineRunner().run(task)


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

Any Pipecat transport takes the filter the same way, as audio_in_filter= in its TransportParams. Push FilterEnableFrame(False) to switch it off mid-call (the stream closes and audio passes through) and FilterEnableFrame(True) to switch it back on. Before you turn it on:

Latency. Enhancement runs on KugelAudio’s servers, not on the device, so the audio makes a round trip over the network and the bot hears the user somewhat later than with an on-device filter. Frames keep their size; they arrive delayed.
Plan and limits. Enhancement uses your organization’s Clarity plan: the free month, then paid. Every second of input audio is enhanced, silence included, and each running pipeline holds one stream open, so on the free month its concurrency limit caps how many sessions can be enhanced at once. See Free month and paid plan.
Connection. The filter opens its connection when the pipeline starts, before the first audio, and closes it when the pipeline stops. After 60 seconds without audio the server closes it and the next audio reconnects.
Failures. While the stream connects, or reconnects after a rate limit or a network error, the bot hears the original audio and a warning is logged. A rejected API key, spent credits, an ended free month or an invalid request turn enhancement off for the rest of the session with one error log; pass on_error= to be told.

Parameter	Default	Description
`api_key`	`KUGELAUDIO_API_KEY` env	API key.
`client`	`None`	An existing `KugelAudio` client, instead of `api_key` and `base_url`.
`model`	`"clarity-1"`	Enhancement model.
`speaker`	`None`	`load_audio(...)` sample of the one voice to keep. Unset: remove noise, keep every voice.
`max_backlog_s`	`2.0`	Seconds the server may fall behind before the original audio is used and the stream reconnects.
`on_error`	`None`	Called once with the error that turned enhancement off.
`base_url`	`https://api.kugelaudio.com`	API base URL.

Environment Variables

Variable	Required	Description
`KUGELAUDIO_API_KEY`	Yes	Your KugelAudio API key
`DEEPGRAM_API_KEY`	Yes*	Required if using Deepgram STT
`OPENAI_API_KEY`	Yes*	Required if using OpenAI LLM

Troubleshooting

API key not found

Make sure KUGELAUDIO_API_KEY is set in your environment or pass api_key directly:

tts = KugelAudioTTSService(api_key="your-api-key", voice_id=1071)

Unsupported sample rate error

The service accepts 24000, 22050, 16000 and 8000. Set the same rate on the service and on your transport’s audio output (in the Quick Start, sample_rate=24000 and audio_out_sample_rate=24000).

WebSocket connection fails

Verify your base_url is correct and the KugelAudio API is reachable. The service connects via WebSocket (wss://) for audio streaming. If a persistent connection drops, the service automatically reconnects on the next run_tts() call.

High latency

Check Performance first: language unset and prewarm() not called (or called outside the event loop) are the usual causes. Then measure from the same region as the endpoint, see Measuring TTFA correctly.

Python version incompatibility

The Pipecat integration requires Python 3.11 or newer. Check your version:

python --version

Next Steps

LiveKit Integration

Use KugelAudio with LiveKit Agents

Speech enhancement

Free month, paid plan and limits for the noise filter

Voice prompting

Prompt patterns that make an LLM sound good when spoken

Get started

Build a voice agent

Generate audio

Voices

Enhance audio

Hosting and deployment

SDK Reference

API Reference

Why Use KugelAudio with Pipecat?

Installation

Quick Start

Configuration

Service Parameters

Supported Sample Rates

Performance

Usage Patterns

Updating the Voice at Runtime

Pipeline Frame Flow

Metrics Support

Removing Background Noise

Environment Variables

Troubleshooting

Next Steps

LiveKit Integration

Speech enhancement

Voice prompting

​Why Use KugelAudio with Pipecat?

​Installation

​Quick Start

​Configuration

​Service Parameters

​Supported Sample Rates

​Performance

​Usage Patterns

​Updating the Voice at Runtime

​Pipeline Frame Flow

​Metrics Support

​Removing Background Noise

​Environment Variables

​Troubleshooting

​Next Steps

LiveKit Integration

Speech enhancement

Voice prompting

Why Use KugelAudio with Pipecat?

Installation

Quick Start

Configuration

Service Parameters

Supported Sample Rates

Performance

Usage Patterns

Updating the Voice at Runtime

Pipeline Frame Flow

Metrics Support

Removing Background Noise

Environment Variables

Troubleshooting

Next Steps