feat(tts): add xAI as a Voice (TTS) provider (#476)

* feat(tts): add xAI as a Voice (TTS) provider

pipecat already ships an xAI TTS service (XAITTSService, WebSocket
streaming) but dograh never wired it into the service configuration, so
xAI could not be selected as a Voice provider in the cascading pipeline.

Wire it through:
- registry: ServiceProviders.XAI + XAITTSConfiguration (voices
  eve/ara/leo/rex/sal, language, computed model) registered in TTSConfig
- service_factory: build XAITTSService in create_tts_service
- check_validity: api-key validation hook
- tests for the factory + docs

The Voice provider dropdown is schema-driven, so xAI appears with no UI
changes.

* fix(tts): validate xAI API key and drop misleading auto-language hint

Addresses review feedback on the xAI Voice provider:

- check_validity: replace the no-op xAI key check with real validation
  against xAI's OpenAI-compatible API (models.list on https://api.x.ai/v1),
  so a bad BYOK key is caught at configuration time instead of at call time.
- registry: remove the "auto" language hint from the field description.
  pipecat's Language enum has no "auto" member, so the factory fell back to
  English silently; the description no longer advertises detection we don't do.
- tests: cover xAI key validation (registered, accepts valid, rejects bad).

* fix(tts): validate xAI key against the TTS voices endpoint

xAI supports endpoint-scoped API keys, so a key scoped to Text-to-Speech
may lack the /v1/models ACL and would be wrongly rejected by the previous
models.list() check. Validate against GET /v1/tts/voices instead — the
scope the key actually needs for TTS — treating 401/403 as an invalid key
and connection errors as a clean, actionable message.

* fix: harden xAI TTS integration

---------

Co-authored-by: Sabiha Khan <sabihak89@gmail.com>
Co-authored-by: Sabiha Khan <87858386+chewwbaka@users.noreply.github.com>

api/services/configuration/check_validity.py

@@ -64,6 +64,7 @@ class UserConfigurationValidator:
             ServiceProviders.RIME.value: self._check_rime_api_key,
             ServiceProviders.MINIMAX.value: self._check_minimax_api_key,
             ServiceProviders.SMALLEST.value: self._check_smallest_api_key,
+            ServiceProviders.XAI.value: self._check_xai_api_key,
         }
 
     async def validate(
@@ -376,6 +377,32 @@ class UserConfigurationValidator:
     def _check_grok_realtime_api_key(self, model: str, api_key: str) -> bool:
         return True
 
+    def _check_xai_api_key(self, model: str, api_key: str) -> bool:
+        # Use the TTS voices endpoint as a best-effort smoke test. Some xAI keys
+        # can be scoped in ways that block listing voices even though the key is
+        # still intended for TTS usage, so only a clear auth failure rejects save.
+        try:
+            response = httpx.get(
+                "https://api.x.ai/v1/tts/voices",
+                headers={"Authorization": f"Bearer {api_key}"},
+                timeout=10.0,
+            )
+        except httpx.RequestError:
+            raise ValueError(
+                "Could not connect to the xAI API. Please check your network "
+                "connection and try again."
+            )
+        if response.status_code == 200:
+            return True
+        if response.status_code == 401:
+            raise ValueError(
+                "Invalid xAI API key. The key was rejected by the xAI API. "
+                "Please check that your API key is correct and active. "
+                "You can verify your keys at "
+                "https://console.x.ai."
+            )
+        return True
+
     def _check_ultravox_realtime_api_key(self, model: str, api_key: str) -> bool:
         return True
 

api/services/configuration/registry.py

@@ -92,6 +92,7 @@ class ServiceProviders(str, Enum):
     GOOGLE_VERTEX_REALTIME = "google_vertex_realtime"
     AZURE_REALTIME = "azure_realtime"
     SMALLEST = "smallest"
+    XAI = "xai"
 
 
 class BaseServiceConfiguration(BaseModel):
@@ -122,6 +123,7 @@ class BaseServiceConfiguration(BaseModel):
         ServiceProviders.AZURE_REALTIME,
         ServiceProviders.SARVAM,
         ServiceProviders.SMALLEST,
+        ServiceProviders.XAI,
     ]
     api_key: str | list[str]
 
@@ -256,6 +258,7 @@ GOOGLE_VERTEX_REALTIME_PROVIDER_MODEL_CONFIG = provider_model_config(
 DEEPGRAM_PROVIDER_MODEL_CONFIG = provider_model_config("Deepgram")
 ELEVENLABS_PROVIDER_MODEL_CONFIG = provider_model_config("ElevenLabs")
 CARTESIA_PROVIDER_MODEL_CONFIG = provider_model_config("Cartesia")
+XAI_PROVIDER_MODEL_CONFIG = provider_model_config("xAI")
 INWORLD_PROVIDER_MODEL_CONFIG = provider_model_config(
     "Inworld",
     description=(
@@ -1278,6 +1281,32 @@ class SmallestAITTSConfiguration(BaseTTSConfiguration):
     )
 
 
+XAI_TTS_VOICES = ["eve", "ara", "leo", "rex", "sal"]
+
+
+@register_tts
+class XAITTSConfiguration(BaseServiceConfiguration):
+    model_config = XAI_PROVIDER_MODEL_CONFIG
+    provider: Literal[ServiceProviders.XAI] = ServiceProviders.XAI
+    voice: str = Field(
+        default="eve",
+        description="xAI voice persona.",
+        json_schema_extra={"examples": XAI_TTS_VOICES, "allow_custom_input": True},
+    )
+    language: str = Field(
+        default="en",
+        description="BCP-47 language code for synthesis (e.g. 'en', 'fr', 'de'), or 'auto' for automatic language detection.",
+        json_schema_extra={"allow_custom_input": True},
+    )
+
+    @computed_field
+    @property
+    def model(self) -> str:
+        # xAI TTS has no separate model selector; the voice fully specifies the
+        # output. A constant keeps the shared `.model` contract satisfied.
+        return "xai-tts"
+
+
 TTSConfig = Annotated[
     Union[
         DeepgramTTSConfiguration,
@@ -1294,6 +1323,7 @@ TTSConfig = Annotated[
         MiniMaxTTSConfiguration,
         AzureSpeechTTSConfiguration,
         SmallestAITTSConfiguration,
+        XAITTSConfiguration,
     ],
     Field(discriminator="provider"),
 ]

api/services/pipecat/service_factory.py

@@ -81,6 +81,7 @@ from pipecat.services.speechmatics.stt import (
     SpeechmaticsSTTService,
     SpeechmaticsSTTSettings,
 )
+from pipecat.services.xai.tts import XAIHttpTTSService, XAITTSSettings
 from pipecat.transcriptions.language import Language
 from pipecat.utils.text.xml_function_tag_filter import XMLFunctionTagFilter
 
@@ -740,6 +741,28 @@ def create_tts_service(
             skip_aggregator_types=["recording_router", "recording"],
             silence_time_s=1.0,
         )
+    elif user_config.tts.provider == ServiceProviders.XAI.value:
+        voice = getattr(user_config.tts, "voice", None) or "eve"
+        language_code = getattr(user_config.tts, "language", None) or "en"
+        if language_code.lower() == "auto":
+            pipecat_language = "auto"
+        else:
+            try:
+                pipecat_language = Language(language_code)
+            except ValueError:
+                pipecat_language = Language.EN
+        return XAIHttpTTSService(
+            api_key=user_config.tts.api_key,
+            sample_rate=audio_config.transport_out_sample_rate,
+            encoding="pcm",
+            settings=XAITTSSettings(
+                voice=voice,
+                language=pipecat_language,
+            ),
+            text_filters=[xml_function_tag_filter],
+            skip_aggregator_types=["recording_router", "recording"],
+            silence_time_s=1.0,
+        )
     else:
         raise HTTPException(
             status_code=400, detail=f"Invalid TTS provider {user_config.tts.provider}"

api/tests/test_xai_tts_service_factory.py

@@ -0,0 +1,160 @@
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+from pipecat.transcriptions.language import Language
+
+from api.services.configuration.check_validity import UserConfigurationValidator
+from api.services.configuration.registry import (
+    XAI_TTS_VOICES,
+    ServiceProviders,
+    XAITTSConfiguration,
+)
+from api.services.pipecat.service_factory import create_tts_service
+
+
+def test_xai_tts_configuration_defaults():
+    config = XAITTSConfiguration(api_key="test-key")
+
+    assert config.provider == ServiceProviders.XAI
+    assert config.voice == "eve"
+    assert config.language == "en"
+    # xAI TTS has no model selector; a constant satisfies the shared contract.
+    assert config.model == "xai-tts"
+    assert XAI_TTS_VOICES == ["eve", "ara", "leo", "rex", "sal"]
+
+
+@pytest.mark.parametrize("transport_out_sample_rate", [8000, 16000])
+def test_create_xai_tts_service_uses_pipeline_compatible_audio_format(
+    transport_out_sample_rate,
+):
+    user_config = SimpleNamespace(
+        tts=SimpleNamespace(
+            provider=ServiceProviders.XAI.value,
+            api_key="test-key",
+            model="xai-tts",
+            voice="rex",
+            language="en",
+        )
+    )
+    audio_config = SimpleNamespace(
+        transport_out_sample_rate=transport_out_sample_rate,
+        transport_in_sample_rate=16000,
+    )
+
+    with patch("api.services.pipecat.service_factory.XAIHttpTTSService") as mock_service:
+        create_tts_service(user_config, audio_config)
+
+    assert mock_service.call_count == 1
+    kwargs = mock_service.call_args.kwargs
+    assert kwargs["api_key"] == "test-key"
+    assert kwargs["sample_rate"] == transport_out_sample_rate
+    assert kwargs["encoding"] == "pcm"
+    assert kwargs["settings"].voice == "rex"
+    assert kwargs["settings"].language == Language.EN
+
+
+def test_create_xai_tts_service_converts_language():
+    user_config = SimpleNamespace(
+        tts=SimpleNamespace(
+            provider=ServiceProviders.XAI.value,
+            api_key="test-key",
+            model="xai-tts",
+            voice="eve",
+            language="fr",
+        )
+    )
+    audio_config = SimpleNamespace(
+        transport_out_sample_rate=24000,
+        transport_in_sample_rate=16000,
+    )
+
+    with patch("api.services.pipecat.service_factory.XAIHttpTTSService") as mock_service:
+        create_tts_service(user_config, audio_config)
+
+    kwargs = mock_service.call_args.kwargs
+    assert kwargs["settings"].language == Language.FR
+
+
+def test_create_xai_tts_service_falls_back_to_english_for_unknown_language():
+    user_config = SimpleNamespace(
+        tts=SimpleNamespace(
+            provider=ServiceProviders.XAI.value,
+            api_key="test-key",
+            model="xai-tts",
+            voice="eve",
+            language="not-a-language",
+        )
+    )
+    audio_config = SimpleNamespace(
+        transport_out_sample_rate=24000,
+        transport_in_sample_rate=16000,
+    )
+
+    with patch("api.services.pipecat.service_factory.XAIHttpTTSService") as mock_service:
+        create_tts_service(user_config, audio_config)
+
+    kwargs = mock_service.call_args.kwargs
+    assert kwargs["settings"].language == Language.EN
+
+
+def test_create_xai_tts_service_preserves_auto_language():
+    user_config = SimpleNamespace(
+        tts=SimpleNamespace(
+            provider=ServiceProviders.XAI.value,
+            api_key="test-key",
+            model="xai-tts",
+            voice="eve",
+            language="auto",
+        )
+    )
+    audio_config = SimpleNamespace(
+        transport_out_sample_rate=24000,
+        transport_in_sample_rate=16000,
+    )
+
+    with patch("api.services.pipecat.service_factory.XAIHttpTTSService") as mock_service:
+        create_tts_service(user_config, audio_config)
+
+    kwargs = mock_service.call_args.kwargs
+    assert kwargs["settings"].language == "auto"
+
+
+def test_xai_is_registered_for_key_validation():
+    validator = UserConfigurationValidator()
+    assert ServiceProviders.XAI.value in validator._validator_map
+
+
+def test_xai_key_validation_accepts_valid_key():
+    validator = UserConfigurationValidator()
+    with patch(
+        "api.services.configuration.check_validity.httpx.get"
+    ) as mock_get:
+        mock_get.return_value.status_code = 200
+        assert validator._check_xai_api_key("xai", "xai-valid-key") is True
+    # Validates against the TTS-scoped voices endpoint, not /v1/models.
+    called_url = mock_get.call_args.args[0]
+    assert called_url == "https://api.x.ai/v1/tts/voices"
+    assert (
+        mock_get.call_args.kwargs["headers"]["Authorization"]
+        == "Bearer xai-valid-key"
+    )
+
+
+def test_xai_key_validation_rejects_bad_key():
+    validator = UserConfigurationValidator()
+    with patch(
+        "api.services.configuration.check_validity.httpx.get"
+    ) as mock_get:
+        mock_get.return_value.status_code = 401
+        with pytest.raises(ValueError):
+            validator._check_xai_api_key("xai", "bad-key")
+
+
+def test_xai_key_validation_allows_scoped_key_without_voice_list_access():
+    validator = UserConfigurationValidator()
+    with patch(
+        "api.services.configuration.check_validity.httpx.get"
+    ) as mock_get:
+        mock_get.return_value.status_code = 403
+        assert validator._check_xai_api_key("xai", "tts-scoped-key") is True