VSIP Python SDK

Real-time voice events for AI voice agents: semantic turn detection, identity-gated barge-in, and speaker verification over one WebSocket.

pip install vsip-sdk            # once published; for now: pip install -e sdk/python

60-second integration

import asyncio
from vsip import VSIPClient, VoiceSession, Turn, int16_to_wav

client = VSIPClient(api_key="vsip_...", url="wss://api.vsip.online/v1/stream")

async def on_turn(turn: Turn):
    if not turn.should_respond:
        return                                  # verified: not your enrolled user
    text = await my_stt(int16_to_wav(turn.audio))   # STT-ready WAV slice
    reply = await my_llm(text, interrupted=turn.interrupted_agent)
    await session.speak_gate()                  # never talk over an open turn
    await client.agent_speaking()               # arm barge-in detection
    await my_tts_play(reply)                    # stop this in on_barge_in
    await client.agent_idle()

def on_barge_in(event):
    my_tts_stop()                               # session already sent agent_idle

session = VoiceSession(client, on_turn=on_turn, on_barge_in=on_barge_in)

async def main():
    async with client:
        await session.start()
        async for frame in my_mic_frames():     # 16kHz mono int16 PCM
            await session.send_audio(frame)

asyncio.run(main())

What VoiceSession does for you

These behaviors are the recommended protocol usage (each one exists because skipping it caused a real failure in field testing):

Lower level

VSIPClient alone gives you: typed events (async for event in client.events()), control commands (agent_speaking/idle, set_barge_in_mode, enroll_speaker, toggle_lock, reset), tagged frames for aec="server" (send_mic_frame / send_tts_reference), and transparent reconnect+resume (sessions survive network blips ≤60s — enrollment and lock state included).

Unknown event types parse as GenericEvent and unknown fields are ignored: the wire protocol is additive-only, so this SDK won't break as the API grows.

Pipecat adapter

pip install "vsip-sdk[pipecat]"

from vsip import VSIPClient
from vsip.adapters.pipecat import VSIPProcessor

vsip = VSIPProcessor(VSIPClient(api_key="vsip_...", url="wss://.../v1/stream"))

pipeline = Pipeline([
    transport.input(),
    stt,
    vsip,          # identity-gated interruption + verification gating
    context_aggr,
    llm,
    tts,
    transport.output(),
])

• Identity-gated interruption: VSIP's barge_in becomes a pipeline interruption — only your enrolled speaker can cut the bot off. Set allow_interruptions=False on the transport so VSIP is the sole authority.
• Verification gating: once locked, TranscriptionFrames are held until the speaker_verification verdict — impostor turns never reach your LLM. Each verdict is also emitted as a VSIPVerificationFrame.
• Automatic TTS sync: Bot speaking-state frames drive set_agent_state.

LiveKit Agents plugin

pip install "vsip-sdk[livekit]"

from livekit.agents import AgentSession
from vsip import VSIPClient
from vsip.adapters.livekit import VSIPVAD

vsip_vad = VSIPVAD(
    VSIPClient(api_key="vsip_...", url="wss://.../v1/stream"),
    on_verification=lambda v: None,   # optional: gate the agent's response
)

session = AgentSession(vad=vsip_vad, stt=..., llm=..., tts=...)

VSIP plugs in as the session's VAD / turn detector: turn_start and verified barge_in become START_OF_SPEECH, turn_end becomes END_OF_SPEECH. So VSIP's semantic endpointing and identity-gated barge-in drive the agent's turn-taking — a cough or bystander can't interrupt. Use the on_verification callback to additionally gate the agent's response on should_respond.

Twilio Media Streams bridge

Connect a phone call's audio to VSIP — VSIP ingests Twilio's μ-law natively. The bridge is framework-agnostic (no extra dependency); wire it to your FastAPI / Starlette / websockets endpoint:

from vsip import VSIPClient
from vsip.adapters.twilio import TwilioBridge

@app.websocket("/twilio-stream")
async def twilio_stream(ws):
    await ws.accept()
    client = VSIPClient(api_key="vsip_...", url="wss://.../v1/stream",
                        audio_format="mulaw8k")
    bridge = TwilioBridge(client,
                          on_turn_end=lambda e: ...,
                          on_verification=lambda v: ...)   # gate on should_respond
    async with client:
        await bridge.run(ws.iter_text(), send=ws.send_text)

On barge_in the bridge sends Twilio a clear message — flushing buffered playback so the caller's interruption stops your TTS instantly.

Availability

Examples

• examples/echo_agent.py — minimal end-to-end agent loop against a local API with your microphone.