"""LiveKit Agents TTS plugin for the Comsync TTS API (run11, Hindi / Hinglish / English). from livekit_comsync import TTS session = AgentSession( stt=..., llm=..., tts=TTS(api_key=os.environ["COMSYNC_TTS_API_KEY"], voice="siya_lively_best"), ) Streaming mode (default) opens one WebSocket per agent reply: LLM text is forwarded as it is generated, the server phrases it and streams 24 kHz PCM back as soon as the first phrase is ready. Interrupting the agent closes the socket, which stops generation on the server immediately. Needs livekit-agents >= 1.2 and aiohttp. Config can also come from COMSYNC_TTS_API_KEY / COMSYNC_TTS_BASE_URL. """ from __future__ import annotations import asyncio import json import os from dataclasses import dataclass, replace import aiohttp from livekit.agents import APIConnectionError, APIConnectOptions, APIError, APIStatusError, APITimeoutError, tts, utils from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS, NOT_GIVEN, NotGivenOr from livekit.agents.utils import is_given DEFAULT_BASE_URL = "https://api.trycomsync.com" SAMPLE_RATE = 24000 @dataclass class _Options: voice: str temperature: float speed: float class TTS(tts.TTS): def __init__( self, *, api_key: str | None = None, base_url: str | None = None, voice: str = "siya_lively_best", temperature: float = 0.3, speed: float = 1.0, http_session: aiohttp.ClientSession | None = None, ) -> None: super().__init__(capabilities=tts.TTSCapabilities(streaming=True), sample_rate=SAMPLE_RATE, num_channels=1) self._api_key = api_key or os.environ.get("COMSYNC_TTS_API_KEY") if not self._api_key: raise ValueError("Comsync TTS API key is required (api_key= or COMSYNC_TTS_API_KEY)") self._base_url = (base_url or os.environ.get("COMSYNC_TTS_BASE_URL") or DEFAULT_BASE_URL).rstrip("/") self._opts = _Options(voice=voice, temperature=temperature, speed=speed) self._session = http_session @property def model(self) -> str: return "comsync-run11" @property def provider(self) -> str: return "comsync" def update_options(self, *, voice: NotGivenOr[str] = NOT_GIVEN, temperature: NotGivenOr[float] = NOT_GIVEN, speed: NotGivenOr[float] = NOT_GIVEN) -> None: if is_given(voice): self._opts.voice = voice if is_given(temperature): self._opts.temperature = temperature if is_given(speed): self._opts.speed = speed def _http(self) -> aiohttp.ClientSession: if self._session is None: self._session = utils.http_context.http_session() return self._session def _headers(self) -> dict: return {"Authorization": f"Bearer {self._api_key}"} def synthesize(self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS) -> ChunkedStream: return ChunkedStream(tts=self, input_text=text, conn_options=conn_options) def stream(self, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS) -> SynthesizeStream: return SynthesizeStream(tts=self, conn_options=conn_options) class ChunkedStream(tts.ChunkedStream): """One-shot synthesis over HTTP (used by LiveKit for non-streaming paths, e.g. session.say()).""" def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None: super().__init__(tts=tts, input_text=input_text, conn_options=conn_options) self._tts: TTS = tts self._opts = replace(tts._opts) async def _run(self, output_emitter: tts.AudioEmitter) -> None: output_emitter.initialize(request_id=utils.shortuuid(), sample_rate=SAMPLE_RATE, num_channels=1, mime_type="audio/pcm") body = {"input": self._input_text, "voice": self._opts.voice, "temperature": self._opts.temperature, "speed": self._opts.speed, "response_format": "pcm"} try: async with self._tts._http().post( f"{self._tts._base_url}/v1/audio/speech", headers=self._tts._headers(), json=body, timeout=aiohttp.ClientTimeout(total=120, sock_connect=self._conn_options.timeout), ) as resp: if resp.status != 200: raise APIStatusError(await resp.text(), status_code=resp.status, request_id=None, body=None) async for data, _ in resp.content.iter_chunks(): output_emitter.push(data) output_emitter.flush() except asyncio.TimeoutError: raise APITimeoutError() from None except aiohttp.ClientError as e: raise APIConnectionError() from e class SynthesizeStream(tts.SynthesizeStream): """Text-in / audio-out over the /v1/stream WebSocket: one socket and one segment per agent reply.""" def __init__(self, *, tts: TTS, conn_options: APIConnectOptions) -> None: super().__init__(tts=tts, conn_options=conn_options) self._tts: TTS = tts self._opts = replace(tts._opts) async def _run(self, output_emitter: tts.AudioEmitter) -> None: request_id = utils.shortuuid() output_emitter.initialize(request_id=request_id, sample_rate=SAMPLE_RATE, num_channels=1, mime_type="audio/pcm", stream=True) output_emitter.start_segment(segment_id=request_id) url = self._tts._base_url.replace("https://", "wss://").replace("http://", "ws://") + "/v1/stream" try: ws = await asyncio.wait_for(self._tts._http().ws_connect(url, headers=self._tts._headers(), heartbeat=20), self._conn_options.timeout) except asyncio.TimeoutError: raise APITimeoutError() from None except aiohttp.WSServerHandshakeError as e: raise APIStatusError(f"Comsync TTS rejected the connection: {e.message}", status_code=e.status, request_id=request_id, body=None) from e except aiohttp.ClientError as e: raise APIConnectionError() from e counts = {"flushes": 0, "done": 0, "input_done": False} async def send() -> None: await ws.send_str(json.dumps({"type": "config", "voice": self._opts.voice, "temperature": self._opts.temperature, "speed": self._opts.speed})) pending_text = False async for data in self._input_ch: if isinstance(data, self._FlushSentinel): if pending_text: await ws.send_str(json.dumps({"type": "flush"})) counts["flushes"] += 1 pending_text = False continue self._mark_started() await ws.send_str(json.dumps({"type": "text", "text": data})) pending_text = True if pending_text: await ws.send_str(json.dumps({"type": "flush"})) counts["flushes"] += 1 counts["input_done"] = True if counts["done"] >= counts["flushes"]: await ws.close() async def recv() -> None: while True: msg = await ws.receive() if msg.type == aiohttp.WSMsgType.BINARY: output_emitter.push(msg.data) elif msg.type == aiohttp.WSMsgType.TEXT: ev = json.loads(msg.data) if ev.get("type") == "done": counts["done"] += 1 if counts["input_done"] and counts["done"] >= counts["flushes"]: return elif ev.get("type") == "error": raise APIError(f"Comsync TTS error: {ev.get('message')}") elif msg.type in (aiohttp.WSMsgType.CLOSE, aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.CLOSING): if counts["input_done"] and counts["done"] >= counts["flushes"]: return raise APIStatusError("Comsync TTS connection closed unexpectedly", status_code=ws.close_code or -1, request_id=request_id, body=None) tasks = [asyncio.create_task(send()), asyncio.create_task(recv())] try: await asyncio.gather(*tasks) finally: await utils.aio.gracefully_cancel(*tasks) await ws.close() output_emitter.end_segment()