can I use WebSocket STT and if yes then how can i implement in python sdk

Thanks for reaching out, Darshak. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I want to use TryHamsa STT as websocket below is smaple code now how can i integrate into livekit python sdk like baseten STT ?

async with websockets.connect(ws_url) as ws:
        print("Connected to WebSocket")

        # Send handshake
        handshake = {
            "type": "handshake",
            "api_key": api_key,
            "options": {
                "silence_timeout": 30.0,
                "sample_rate": sample_rate,
                "min_silence_duration_ms": 50,
                "min_speech_ms": 50,
                "vad_threshold": 0.6,
                "eos_enabled": True,
                "eos_threshold": 0.6,
                "client_logging": True
            }
        }

        await ws.send(json.dumps(handshake))
        ack = await ws.recv()
        ack_data = json.loads(ack)

        if ack_data.get("type") != "handshake_ack":
            print(f"Auth failed: {ack_data.get('error')}")
            return

        print("✓ Connected and authenticated")
        print("Streaming audio...")

        # Run send and receive concurrently
        send_task = asyncio.create_task(
            send_wav_chunks(ws, wf, frames_per_chunk, sample_rate)
        )
        receive_task = asyncio.create_task(receive_results(ws))

        await asyncio.gather(send_task, receive_task)


async def send_wav_chunks(ws, wf, frames_per_chunk, sample_rate):
    """Read and send WAV chunks to the server"""
    try:
        while True:
            frames = wf.readframes(frames_per_chunk)
            if not frames:
                print("\nFinished streaming audio.")
                await ws.send(json.dumps({"type": "end_of_stream"}))
                break

            await ws.send(frames)
            await asyncio.sleep(frames_per_chunk / sample_rate)  # simulate real-time pacing
    except Exception as e:
        print(f"Send error: {e}")


async def receive_results(ws):
    """Receive transcription/diarization results"""
    try:
        while True:
            msg = await ws.recv()
            data = json.loads(msg)

            if data.get("type") == "transcription":
                result = data.get("data", {})
                print(f"\n data: {data}")
                print(f"\n✓ {result.get('transcription')}")
                print(f"  Gender: {result.get('gender')} | EOS: {result.get('eos', {}).get('prediction'):.2f}")

            elif data.get("type") == "log":
                print(f"  [Log] {data.get('data')}")

            elif data.get("type") == "error":
                print(f"\n✗ Error: {data.get('error')}")
                break
    except Exception as e:
        print(f"Receive error: {e}")

Thanks — I see you want to use TryHamsa’s WebSocket STT (your sample) inside LiveKit the same way Baseten is used. You can do that by wrapping your WebSocket flow in a LiveKit-compatible STT class and using it in an AgentSession or as a standalone stream. Below is a minimal, focused plan plus a compact template you can drop into your code. Plan (what to do) • Implement a custom STT class (e.g.,

TryHamsaSTT

) that implements LiveKit’s `SpeechToTextEngine`/`SpeechStream` interface. • Reuse your handshake/send/receive logic inside that stream class. • Implement

push_frame(frame)

to send audio bytes to TryHamsa and make the stream yield LiveKit `SpeechEvent`s for interim/final transcripts. • Use the class in

AgentSession(stt=TryHamsaSTT(...))

or call

stream()

directly for standalone usage. Minimal template (adapt your sample into this)

import asyncio, json, websockets, logging
from livekit.agents import stt
from livekit import rtc

class TryHamsaSTT(stt.SpeechToTextEngine):
    def __init__(self, api_key: str, ws_url: str, sample_rate: int = 16000):
        self.api_key = api_key
        self.ws_url = ws_url
        self.sample_rate = sample_rate
        super().__init__()

    def stream(self, *, language: str = "", cancellation_event: asyncio.Event = None):
        return TryHamsaStream(self.api_key, self.ws_url, self.sample_rate, cancellation_event)


class TryHamsaStream(stt.SpeechStream):
    def __init__(self, api_key, ws_url, sample_rate, cancellation_event=None):
        self.api_key = api_key
        self.ws_url = ws_url
        self.sample_rate = sample_rate
        self.cancellation_event = cancellation_event or asyncio.Event()
        self._ws = None
        self._queue = asyncio.Queue()
        self._closed = False

    async def _connect(self):
        self._ws = await websockets.connect(self.ws_url)
        handshake = {
            "type": "handshake",
            "api_key": self.api_key,
            "options": {"sample_rate": self.sample_rate}
        }
        await self._ws.send(json.dumps(handshake))
        ack = json.loads(await self._ws.recv())
        if ack.get("type") != "handshake_ack":
            raise Exception(f"Auth failed: {ack.get('error')}")
        asyncio.create_task(self._receive_results())

    async def _receive_results(self):
        try:
            async for msg in self._ws:
                if self.cancellation_event.is_set():
                    break
                data = json.loads(msg)
                if data.get("type") == "transcription":
                    result = data.get("data", {})
                    text = result.get("transcription", "")
                    is_final = result.get("eos", {}).get("final", False)
                    event_type = stt.SpeechEventType.FINAL_TRANSCRIPT if is_final else stt.SpeechEventType.INTERIM_TRANSCRIPT
                    await self._queue.put(stt.SpeechEvent(
                        type=event_type,
                        alternatives=[stt.SpeechEventAlternative(text=text, confidence=1.0)]
                    ))
                elif data.get("type") == "end_of_speech":
                    await self._queue.put(stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH))
        except Exception as e:
            await self._queue.put(stt.SpeechEvent(type=stt.SpeechEventType.ERROR, error=e))
        finally:
            await self.close()

    async def push_frame(self, frame: rtc.AudioFrame):
        if not self._ws:
            await self._connect()
        try:
            # LiveKit AudioFrame data is LPCM16; send raw bytes as your sample did
            await self._ws.send(frame.data)
        except Exception as e:
            logging.error("Send error: %s", e)

    async def flush(self):
        if self._ws and not self._closed:
            await self._ws.send(json.dumps({"type": "end_of_stream"}))

    async def close(self):
        self._closed = True
        if self._ws:
            await self._ws.close()

    async def __aiter__(self):
        if not self._ws:
            await self._connect()
        while not self._closed:
            try:
                item = await asyncio.wait_for(self._queue.get(), timeout=1.0)
                yield item
            except asyncio.TimeoutError:
                if self.cancellation_event.is_set():
                    break
                continue

How to use • In AgentSession:

from livekit.agents import AgentSession
stt_engine = TryHamsaSTT(api_key="KEY", ws_url="wss://...") 
session = AgentSession(stt=stt_engine, llm="openai/gpt-4o")

• Standalone (pipe audio from a LiveKit track): ◦ Create the stream:

stream = TryHamsaSTT(...).stream()

◦ For each

rtc.AudioFrame

call

await stream.push_frame(frame)

◦ Iterate

async for ev in stream:

to read transcripts. Notes and tips • Ensure frames are in the format TryHamsa expects (mono, 16-bit little-endian PCM). Convert if needed. • Map TryHamsa’s interim/final flags to

SpeechEventType.INTERIM_TRANSCRIPT

and

FINAL_TRANSCRIPT

. • Implement reconnect/error handling and optional

flush()

semantics per your service. • If TryHamsa returns speaker IDs, emit diarization info via custom metadata or additional event fields. If you want, I can: • Convert your exact sample file-reading code (send_wav_chunks/receive_results) into a full working example using LiveKit

rtc.AudioFrame

emulation, or • Scaffold a packaged plugin module named

livekit.plugins.tryhamsa

. Which would you prefer? Sources: Speech-to-text (STT) models | LiveKit Docs | Pipeline nodes and hooks | LiveKit docs | Baseten STT plugin guide | LiveKit docs

suppose i use ws stt then i need to use vad or not or how exact pipeline works in case of websocket stt ?

Good question — I'll explain clearly which parts need VAD, how the pipeline runs with WebSocket STT, and recommended settings. Short answer • You generally should use VAD. • VAD segments speech/silence and signals end-of-utterance (END_OF_SPEECH) so the agent can finalize transcripts and send them to the LLM. • If your STT provider provides reliable real-time endpointing, you can rely on STT endpointing instead, but using both (VAD + STT endpointing) is best practice. Pipeline (step-by-step) 1. Audio capture: LiveKit captures mic audio as LPCM frames. 2. VAD (recommended): A VAD plugin (e.g., Silero VAD) inspects frames and emits START_OF_SPEECH / END_OF_SPEECH. 3. Stream to STT (WebSocket): While VAD reports speech, you stream audio frames over the WebSocket to TryHamsa (or other STT). Streaming can be continuous or chunked. 4. Interim transcripts: STT returns interim results over WS; you can show/process them immediately. 5. Endpointing: When VAD signals END_OF_SPEECH (or the STT signals endpointing), you finalize the current segment: send an `end_of_stream`/`end_of_segment` or stop streaming that utterance so STT returns final transcript. 6. Pass final transcript to LLM / Agent logic. 7. Optionally synthesize response with TTS and play back. When you might skip VAD • If TryHamsa provides low-latency, reliable finalization (explicit final transcripts or EOS events), you can set turn_detection to use STT endpointing (

turn_detection="stt"

). • Skip VAD only if you have confidence in STT endpointing and also handle interruptions and edge cases. Why combining VAD + STT endpointing is better • Faster interruption handling (detect new speaker starting while TTS/LLM runs). • Fallback if STT endpointing lags or misdetects silence. • Cleaner decision points for the agent (when to call the LLM). Practical tips for TryHamsa WebSocket • Keep streaming audio while START_OF_SPEECH is active; on VAD END_OF_SPEECH send your

end_of_stream

(or the provider’s end marker) so TryHamsa emits a final transcription. • Tune VAD parameters (min_silence_duration_ms, vad_threshold) to match your environment. • Use interim transcripts for live feedback; rely on final transcripts for LLM prompts. • Implement short timeout fallback: if neither VAD nor STT EOS appear after N ms of silence, finalize the segment anyway. LiveKit settings examples • VAD-driven turns (recommended): session = AgentSession( stt=TryHamsaSTT(...), vad=silero.VAD.load(), turn_detection="vad", ... ) • STT-driven turns (if provider has endpointing): session = AgentSession( stt=TryHamsaSTT(...), vad=silero.VAD.load(), # still useful as fallback turn_detection="stt", ... ) If you want, I can: • Suggest specific VAD parameter values for TryHamsa based on your expected mic/sample_rate and noise level; or • Show how to wire Silero VAD into the TryHamsa stream class (code snippet) so END_OF_SPEECH triggers sending

{"type":"end_of_stream"}

. Which would you like? Sources: Agents Framework | livekit/livekit_composite | DeepWiki | examples/voice_agents | livekit/livekit_composite | DeepWiki | Turns overview | LiveKit docs