feat(gnani): add X-Source/X-API-Request-ID headers and widen audio co…

livekit-plugins/livekit-plugins-gnani/livekit/plugins/gnani/models.py

@@ -51,6 +51,6 @@
 """``language`` is honored by ``timbre-v2.5`` only. See
 https://docs.gnani.ai/api/TTS/tts-inference#supported-languages"""
 
-GnaniTTSEncodings = Literal["linear_pcm", "oggopus"]
-GnaniTTSContainers = Literal["raw", "mp3", "wav", "ogg"]
-GnaniTTSBitrates = Literal["96k", "128k", "192k"]
+GnaniTTSEncodings = Literal["linear_pcm", "oggopus", "pcm_mulaw", "pcm_alaw"]
+GnaniTTSContainers = Literal["raw", "mp3", "wav", "ogg", "mulaw", "alaw"]
+GnaniTTSBitrates = Literal["32k", "64k", "96k", "128k", "192k"]

livekit-plugins/livekit-plugins-gnani/livekit/plugins/gnani/request_id.py

@@ -0,0 +1,24 @@
+# Copyright 2025 LiveKit, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Request ID generation for Gnani API correlation."""
+
+from __future__ import annotations
+
+import uuid
+
+
+def _generate_request_id() -> str:
+    """Generate a unique request ID for outbound Gnani API calls."""
+    return f"lk_req_{uuid.uuid4().hex[:12]}"

livekit-plugins/livekit-plugins-gnani/livekit/plugins/gnani/stt.py

@@ -48,6 +48,7 @@
 
 from .log import logger
 from .models import GnaniSTTLanguages
+from .request_id import _generate_request_id
 
 GnaniSTTFormat = Literal["verbatim", "transcribe"]
 
@@ -215,8 +216,11 @@ async def _recognize_impl(
         if self._opts.itn_native_numerals:
             form_data.add_field("itn_native_numerals", "true")
 
+        request_id = _generate_request_id()
         headers: dict[str, str] = {
             "X-API-Key-ID": self._opts.api_key,
+            "X-API-Request-ID": request_id,
+            "X-Source": "livekit",
         }
 
         try:
@@ -240,11 +244,11 @@ async def _recognize_impl(
 
                 response_json = await res.json()
                 transcript = response_json.get("transcript", "")
-                request_id = response_json.get("request_id", "")
+                api_request_id = response_json.get("request_id", request_id)
 
                 return stt.SpeechEvent(
                     type=stt.SpeechEventType.FINAL_TRANSCRIPT,
-                    request_id=request_id,
+                    request_id=api_request_id,
                     alternatives=[
                         stt.SpeechData(
                             language=LanguageCode(lang),
@@ -321,10 +325,13 @@ async def _run(self) -> None:
         import websockets
 
         ws_url = self._build_ws_url()
+        ws_request_id = _generate_request_id()
         headers: dict[str, str] = {
             "x-api-key-id": self._opts.api_key,
             "lang_code": self._opts.language,
             "x-sample-rate": str(self._opts.sample_rate),
+            "x-api-request-id": ws_request_id,
+            "x-source": "livekit",
         }
         if self._opts.format != "verbatim":
             headers["x-format"] = self._opts.format

livekit-plugins/livekit-plugins-gnani/livekit/plugins/gnani/tts.py

@@ -55,12 +55,13 @@
     GnaniTTSModels,
     GnaniTTSVoices,
 )
+from .request_id import _generate_request_id
 
 GNANI_TTS_BASE_URL = "https://api.vachana.ai"
 
 GnaniTTSSynthesizeMethod = Literal["rest", "sse", "websocket"]
 
-SUPPORTED_SAMPLE_RATES = (8000, 16000, 22050, 44100)
+SUPPORTED_SAMPLE_RATES = (8000, 16000, 22050, 24000, 44100, 48000)
 
 
 _DEPRECATED_TTS_KWARGS = frozenset(("http_session",))
@@ -239,11 +240,15 @@ def _build_payload(opts: GnaniTTSOptions, text: str) -> dict:
     return payload
 
 
-def _build_headers(opts: GnaniTTSOptions) -> dict[str, str]:
-    return {
+def _build_headers(opts: GnaniTTSOptions, request_id: str | None = None) -> dict[str, str]:
+    headers = {
         "X-API-Key-ID": opts.api_key,
         "Content-Type": "application/json",
+        "X-Source": "livekit",
     }
+    if request_id:
+        headers["X-API-Request-ID"] = request_id
+    return headers
 
 
 def _mime_type(opts: GnaniTTSOptions) -> str:
@@ -266,11 +271,12 @@ def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions
         self._opts = replace(tts._opts)
 
     async def _run(self, output_emitter: tts.AudioEmitter) -> None:
+        api_request_id = _generate_request_id()
         try:
             async with self._tts._ensure_session().post(
                 url=f"{self._opts.base_url}/api/v1/tts/inference",
                 json=_build_payload(self._opts, self._input_text),
-                headers=_build_headers(self._opts),
+                headers=_build_headers(self._opts, request_id=api_request_id),
                 timeout=aiohttp.ClientTimeout(
                     total=self._conn_options.timeout,
                     sock_connect=self._conn_options.timeout,
@@ -326,11 +332,12 @@ def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions
 
     async def _run(self, output_emitter: tts.AudioEmitter) -> None:
         request_id = utils.shortuuid()
+        api_request_id = _generate_request_id()
         try:
             async with self._tts._ensure_session().post(
                 url=f"{self._opts.base_url}/api/v1/tts/sse",
                 json=_build_payload(self._opts, self._input_text),
-                headers=_build_headers(self._opts),
+                headers=_build_headers(self._opts, request_id=api_request_id),
                 timeout=aiohttp.ClientTimeout(
                     total=self._conn_options.timeout,
                     sock_connect=self._conn_options.timeout,
@@ -436,11 +443,12 @@ async def _run(self, output_emitter: tts.AudioEmitter) -> None:
         import websockets
 
         request_id = utils.shortuuid()
+        api_request_id = _generate_request_id()
         try:
             ws_url = self._build_ws_url()
             async with websockets.connect(
                 ws_url,
-                additional_headers=_build_headers(self._opts),
+                additional_headers=_build_headers(self._opts, request_id=api_request_id),
                 ping_interval=20,
                 ping_timeout=20,
                 close_timeout=10,
@@ -547,11 +555,12 @@ async def _run(self, output_emitter: tts.AudioEmitter) -> None:
         segment_id = utils.shortuuid()
         output_emitter.start_segment(segment_id=segment_id)
 
+        api_request_id = _generate_request_id()
         try:
             ws_url = self._build_ws_url()
             async with websockets.connect(
                 ws_url,
-                additional_headers=_build_headers(self._opts),
+                additional_headers=_build_headers(self._opts, request_id=api_request_id),
                 ping_interval=20,
                 ping_timeout=20,
                 close_timeout=10,

tests/test_plugin_gnani_stt.py

@@ -277,3 +277,83 @@ def _fake_create_task(coro, *args, **kwargs):
     with patch("livekit.agents.stt.stt.asyncio.create_task", side_effect=_fake_create_task):
         stream = stt.stream()
     assert stream._opts.language == "hi-IN"
+
+
+async def test_stt_rest_headers_include_source_and_request_id():
+    """REST recognize() sends X-Source: livekit and a generated X-API-Request-ID."""
+    from livekit.agents import APIConnectionError
+    from livekit.plugins.gnani import STT
+
+    class _FakePostCM:
+        async def __aenter__(self):
+            raise APIConnectionError("short-circuit")
+
+        async def __aexit__(self, *exc):
+            return None
+
+    captured: dict = {}
+
+    def _fake_post(url, *, headers=None, data=None, **kwargs):
+        captured.update(url=url, headers=headers)
+        return _FakePostCM()
+
+    fake_session = MagicMock()
+    fake_session.post = _fake_post
+
+    stt = STT(api_key="test-key")
+    stt._session = fake_session
+
+    import numpy as np
+
+    from livekit import rtc
+    from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS
+
+    frame = rtc.AudioFrame(
+        data=np.zeros(1600, dtype=np.int16).tobytes(),
+        sample_rate=16000,
+        num_channels=1,
+        samples_per_channel=1600,
+    )
+
+    with pytest.raises(APIConnectionError):
+        await stt._recognize_impl(buffer=[frame], conn_options=DEFAULT_API_CONNECT_OPTIONS)
+
+    assert captured["headers"]["X-Source"] == "livekit"
+    assert captured["headers"]["X-API-Request-ID"].startswith("lk_req_")
+
+
+async def test_stt_websocket_headers_include_source_and_request_id():
+    """WebSocket streaming sends x-source: livekit and a generated x-api-request-id."""
+    from livekit.agents import APIConnectionError
+    from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS
+    from livekit.plugins.gnani.stt import STT, GnaniSTTOptions, SpeechStream
+
+    captured: dict = {}
+
+    class _FakeConnectCM:
+        async def __aenter__(self):
+            raise ConnectionRefusedError("short-circuit")
+
+        async def __aexit__(self, *exc):
+            return None
+
+    def _fake_connect(url, *, additional_headers=None, **kwargs):
+        captured.update(url=url, headers=additional_headers)
+        return _FakeConnectCM()
+
+    stt = STT(api_key="test-key")
+    opts = GnaniSTTOptions(api_key="test-key", language="en-IN")
+
+    def _fake_create_task(coro, *args, **kwargs):
+        coro.close()
+        return MagicMock()
+
+    with patch("livekit.agents.stt.stt.asyncio.create_task", side_effect=_fake_create_task):
+        stream = SpeechStream(stt=stt, opts=opts, conn_options=DEFAULT_API_CONNECT_OPTIONS)
+
+    with patch("websockets.connect", side_effect=_fake_connect):
+        with pytest.raises(APIConnectionError):
+            await stream._run()
+
+    assert captured["headers"]["x-source"] == "livekit"
+    assert captured["headers"]["x-api-request-id"].startswith("lk_req_")

tests/test_plugin_gnani_tts.py

@@ -363,14 +363,14 @@ def test_tts_rejects_invalid_sample_rate():
     from livekit.plugins.gnani import TTS
 
     with pytest.raises(ValueError, match="sample_rate"):
-        TTS(api_key="test-key", sample_rate=48000)
+        TTS(api_key="test-key", sample_rate=11025)
 
 
 def test_tts_all_sample_rates_accepted():
     """TTS accepts all documented sample rates."""
     from livekit.plugins.gnani import TTS
 
-    for rate in (8000, 16000, 22050, 44100):
+    for rate in (8000, 16000, 22050, 24000, 44100, 48000):
         tts = TTS(api_key="test-key", sample_rate=rate)
         assert tts.sample_rate == rate
 
@@ -421,3 +421,91 @@ def _fake_create_task(coro, *args, **kwargs):
         stream = tts.synthesize("hello")
     assert isinstance(stream, WebSocketChunkedStream)
     assert stream._build_ws_url() == "wss://api.vachana.ai/api/v1/tts"
+
+
+def _patch_session_capture_post(tts, captured: dict) -> None:
+    """Short-circuit ``_ensure_session().post(...)`` right after capturing its
+    kwargs, so the request-building code runs without hitting the network."""
+    from unittest.mock import MagicMock
+
+    from livekit.agents import APIConnectionError
+
+    class _FakePostCM:
+        async def __aenter__(self):
+            raise APIConnectionError("short-circuit")
+
+        async def __aexit__(self, *exc):
+            return None
+
+    def _fake_post(url, *, headers=None, json=None, **kwargs):
+        captured.update(url=url, headers=headers, json=json)
+        return _FakePostCM()
+
+    fake_session = MagicMock()
+    fake_session.post = _fake_post
+    tts._session = fake_session
+
+
+async def test_tts_rest_headers_include_source_and_request_id():
+    """REST synthesis sends X-Source: livekit and a generated X-API-Request-ID."""
+    from livekit.agents import APIConnectionError
+    from livekit.plugins.gnani import TTS
+
+    tts = TTS(api_key="test-key", synthesize_method="rest")
+    captured: dict = {}
+    _patch_session_capture_post(tts, captured)
+
+    with pytest.raises(APIConnectionError):
+        async for _ in tts.synthesize("hello world"):
+            pass
+
+    assert captured["headers"]["X-Source"] == "livekit"
+    assert captured["headers"]["X-API-Request-ID"].startswith("lk_req_")
+
+
+async def test_tts_sse_headers_include_source_and_request_id():
+    """SSE synthesis sends X-Source: livekit and a generated X-API-Request-ID."""
+    from livekit.agents import APIConnectionError
+    from livekit.plugins.gnani import TTS
+
+    tts = TTS(api_key="test-key", synthesize_method="sse")
+    captured: dict = {}
+    _patch_session_capture_post(tts, captured)
+
+    with pytest.raises(APIConnectionError):
+        async for _ in tts.synthesize("hello world"):
+            pass
+
+    assert captured["headers"]["X-Source"] == "livekit"
+    assert captured["headers"]["X-API-Request-ID"].startswith("lk_req_")
+
+
+async def test_tts_websocket_headers_include_source_and_request_id():
+    """WebSocket synthesis sends X-Source: livekit and a generated X-API-Request-ID."""
+    from unittest.mock import patch
+
+    from livekit.agents import APIConnectionError
+    from livekit.plugins.gnani import TTS
+
+    captured: dict = {}
+
+    class _FakeConnectCM:
+        async def __aenter__(self):
+            raise ConnectionRefusedError("short-circuit")
+
+        async def __aexit__(self, *exc):
+            return None
+
+    def _fake_connect(url, *, additional_headers=None, **kwargs):
+        captured.update(url=url, headers=additional_headers)
+        return _FakeConnectCM()
+
+    tts = TTS(api_key="test-key", synthesize_method="websocket")
+
+    with patch("websockets.connect", side_effect=_fake_connect):
+        with pytest.raises(APIConnectionError):
+            async for _ in tts.synthesize("hello world"):
+                pass
+
+    assert captured["headers"]["X-Source"] == "livekit"
+    assert captured["headers"]["X-API-Request-ID"].startswith("lk_req_")