Voice (LiveKit)

Use LiveKit WebRTC for lower-latency, higher-quality voice conversations with built-in acoustic echo cancellation (AEC) and audio processing.

Why LiveKit

The default WebSocket voice transport works well, but LiveKit offers several advantages:

Feature	WebSocket	LiveKit
Latency	Good	Lower (WebRTC)
Echo Cancellation	None (use headphones)	Built-in AEC
Audio Quality	16 kHz PCM	Opus codec, adaptive bitrate
Audio Processing	None	Noise suppression, AGC, high-pass filter

LiveKit is recommended for production applications, especially on devices without headphones where echo cancellation is important.

Prerequisites

Install the LiveKit Python SDK alongside the Estuary SDK:

pip install "estuary-sdk[livekit]"

This installs the livekit package. No additional configuration is needed -- the SDK detects it at runtime.

warning

LiveKit voice requires that your Estuary server has LiveKit enabled and configured. If the server does not support LiveKit, start_voice() will raise an EstuaryError. Use voice_transport="websocket" to bypass LiveKit, or voice_transport="auto" to fall back automatically.

Starting Voice with LiveKit

Set voice_transport to "livekit" in your config:

from estuary_sdk import EstuaryClient, EstuaryConfig

config = EstuaryConfig(
    server_url="https://api.estuary-ai.com",
    api_key="est_your_api_key",
    character_id="your-character-uuid",
    player_id="user-123",
    voice_transport="livekit",
)

async with EstuaryClient(config) as client:
    await client.connect()
    await client.start_voice()

How It Works

When you call start_voice() with the LiveKit transport:

1. The SDK receives a LiveKit token embedded in the session info during connect().
2. The SDK connects to the LiveKit room using the livekit Python SDK.
3. Outgoing audio is processed through the AudioProcessingModule (10ms chunks) with echo cancellation, noise suppression, AGC, and high-pass filter enabled.
4. Incoming bot audio feeds the APM's reverse stream for echo modelling, then is delivered via the audio_received event.
5. Speech-to-text and bot responses still arrive via the Socket.IO connection.

LiveKit Events

async def on_livekit_connected(room):
    print("LiveKit room connected:", room)

async def on_livekit_disconnected():
    print("LiveKit room disconnected")

client.on("livekit_connected", on_livekit_connected)
client.on("livekit_disconnected", on_livekit_disconnected)

Text responses and STT events work the same way as with WebSocket voice:

async def on_stt(response):
    if response.is_final:
        print("You:", response.text)

async def on_response(response):
    if response.is_final:
        print("Bot:", response.text)

client.on("stt_response", on_stt)
client.on("bot_response", on_response)

Audio Received Event

When using LiveKit, decoded bot audio is delivered through the audio_received event as raw bytes:

async def on_audio(audio_bytes):
    # Raw PCM16 audio from the bot -- play or process as needed
    pass

client.on("audio_received", on_audio)

Auto Mode

Set voice_transport="auto" to prefer LiveKit when the livekit package is installed:

config = EstuaryConfig(
    server_url="https://api.estuary-ai.com",
    api_key="est_your_api_key",
    character_id="your-character-uuid",
    player_id="user-123",
    # voice_transport defaults to "websocket"
    voice_transport="auto",
)

await client.start_voice()
# Uses LiveKit if livekit package is installed, otherwise WebSocket

Muting and Stopping

Mute and stop work identically to WebSocket voice:

# Toggle mute
client.toggle_mute()
print("Muted:", client.is_muted)

# Stop voice session
await client.stop_voice()

Error Handling

LiveKit voice can fail at several stages:

from estuary_sdk import EstuaryError, ErrorCode

try:
    await client.start_voice()
except EstuaryError as err:
    if err.code == ErrorCode.LIVEKIT_UNAVAILABLE:
        print("livekit package not installed")
    elif err.code == ErrorCode.CONNECTION_FAILED:
        print("Failed to connect to LiveKit room")
    elif err.code == ErrorCode.VOICE_START_FAILED:
        print("Voice start failed:", err)

Example: LiveKit Voice Chat

import asyncio
from estuary_sdk import EstuaryClient, EstuaryConfig

config = EstuaryConfig(
    server_url="https://api.estuary-ai.com",
    api_key="est_your_api_key",
    character_id="your-character-uuid",
    player_id="user-123",
    voice_transport="livekit",
)

async def main():
    async with EstuaryClient(config) as client:
        async def on_livekit_connected(room):
            print("LiveKit room ready:", room)

        async def on_stt(response):
            if response.is_final:
                print("You:", response.text)

        async def on_response(response):
            if response.is_final:
                print("Bot:", response.text)

        async def on_error(err):
            print("Error:", err)

        client.on("livekit_connected", on_livekit_connected)
        client.on("stt_response", on_stt)
        client.on("bot_response", on_response)
        client.on("error", on_error)

        await client.connect()
        await client.start_voice()
        print("Voice active -- speak into your microphone.")

        await asyncio.Event().wait()

asyncio.run(main())

Next Steps

• Voice (WebSocket) -- Fallback voice transport
• Memory & Knowledge Graph -- Query character memory
• API Reference: EstuaryClient -- Full method reference