Gestion des sessions WebSocket — Text-to-Speech API

Leçon corrigée le 3 novembre 2026 d’après la page Text to Speech de xAI : la documentation parle d’un « permis de session » de 600 secondes et d’une connexion sans délai d’expiration, sans affirmer que le serveur ferme la connexion au bout de 600 secondes depuis son ouverture ; la stratégie de renouvellement est conservée, comme précaution. Source : Text to Speech.

Maîtriser le cycle de vie des sessions

En production, la gestion robuste des sessions WebSocket est ce qui sépare un prototype d’un produit fiable. Les sessions TTS ont des contraintes spécifiques (TTL, limites de concurrence) que votre application doit gérer correctement.

Cycle de vie d’une session

Une session WebSocket TTS suit ce cycle :

Ouverture : connexion à wss://api.x.ai/v1/tts avec les paramètres
Active : échange de messages text.delta/audio.delta
Fermeture normale : le client ou le serveur ferme la connexion
Fermeture imprévue : erreur, maintenance, ou fin du permis de session de 600 secondes que mentionne la documentation

Détection de la fermeture

Le serveur peut fermer la connexion à tout moment (erreur interne, maintenance, fin de permis de session). Votre application doit toujours gérer l’événement de fermeture :

async def managed_session(api_key, language="fr", voice="eve"):
    uri = f"wss://api.x.ai/v1/tts?language={language}&voice={voice}"
    headers = {"Authorization": f"Bearer {api_key}"}

    try:
        async with websockets.connect(uri, additional_headers=headers) as ws:
            yield ws
    except websockets.ConnectionClosed as e:
        print(f"Connexion fermee: code={e.code}, raison={e.reason}")
    except websockets.ConnectionClosedError:
        print("Connexion perdue inopinement")

Reconnexion automatique

Pour les applications longue durée, implémentez un mécanisme de reconnexion avec backoff exponentiel :

import asyncio
import random

class TTSSessionManager:
    def __init__(self, api_key, language="fr", voice="eve"):
        self.api_key = api_key
        self.language = language
        self.voice = voice
        self.ws = None
        self.max_retries = 5

    async def connect(self):
        uri = f"wss://api.x.ai/v1/tts?language={self.language}&voice={self.voice}"
        headers = {"Authorization": f"Bearer {self.api_key}"}

        for attempt in range(self.max_retries):
            try:
                self.ws = await websockets.connect(uri, additional_headers=headers)
                print(f"Connecte (tentative {attempt + 1})")
                return
            except Exception as e:
                wait = min(2 ** attempt + random.random(), 30)
                print(f"Echec connexion: {e}. Retry dans {wait:.1f}s")
                await asyncio.sleep(wait)

        raise ConnectionError("Impossible de se connecter apres 5 tentatives")

    async def ensure_connected(self):
        if self.ws is None or self.ws.closed:
            await self.connect()

    async def synthesize(self, text):
        await self.ensure_connected()

        try:
            await self.ws.send(json.dumps({
                "type": "text.delta",
                "delta": text
            }))
            await self.ws.send(json.dumps({"type": "text.done"}))

            audio_chunks = []
            async for msg in self.ws:
                data = json.loads(msg)
                if data["type"] == "audio.delta":
                    audio_chunks.append(base64.b64decode(data["delta"]))
                elif data["type"] == "audio.done":
                    break
                elif data["type"] == "error":
                    raise Exception(data["message"])

            return b"".join(audio_chunks)

        except websockets.ConnectionClosed:
            # Reconnexion et retry
            await self.connect()
            return await self.synthesize(text)

Gestion de la durée de session

La page Text to Speech de xAI mentionne un « permis de session » de 600 secondes, sans préciser son point de départ ni ce qui se passe à son terme, et indique par ailleurs que la connexion n’a pas de délai d’expiration. L’approche la plus prudente consiste à compter depuis la connexion et à renouveler la session avant ce terme :

import time

class TTLAwareSession:
    TTL = 600  # secondes
    REFRESH_MARGIN = 60  # reconnecter 60s avant expiration

    def __init__(self):
        self.connected_at = None

    async def connect(self):
        # ... connexion ...
        self.connected_at = time.time()

    def should_refresh(self):
        if self.connected_at is None:
            return True
        elapsed = time.time() - self.connected_at
        return elapsed >= (self.TTL - self.REFRESH_MARGIN)

    async def synthesize(self, text):
        if self.should_refresh():
            await self.connect()  # Nouvelle session
        # ... envoi du texte ...

En vous reconnectant 60 secondes avant l’expiration, vous évitez les coupures en pleine synthèse.

Pool de sessions

Pour les applications à fort trafic, maintenez un pool de sessions réutilisables :

class SessionPool:
    def __init__(self, api_key, pool_size=10):
        self.api_key = api_key
        self.pool_size = min(pool_size, 50)  # Max 50 par equipe
        self.available = asyncio.Queue()
        self.active_count = 0

    async def acquire(self):
        if not self.available.empty():
            session = await self.available.get()
            if not session.should_refresh():
                return session

        if self.active_count < self.pool_size:
            session = TTSSessionManager(self.api_key)
            await session.connect()
            self.active_count += 1
            return session

        # Attendre qu'une session se libere
        return await self.available.get()

    async def release(self, session):
        await self.available.put(session)

Limitez la taille du pool à une fraction de la limite de 50 sessions pour laisser de la marge aux autres services de votre application.

Métriques à surveiller

Pour une gestion saine des sessions en production, surveillez :

Nombre de sessions actives : ne doit pas dépasser 50
Taux de reconnexion : un taux élevé indique un problème réseau ou de TTL
Latence du premier audio.delta : mesure la réactivité du service
Erreurs WebSocket : codes de fermeture anormaux (1006, 1011)

class MetricsCollector:
    def __init__(self):
        self.connections = 0
        self.reconnections = 0
        self.errors = 0
        self.first_audio_latencies = []

    def record_connection(self):
        self.connections += 1

    def record_reconnection(self):
        self.reconnections += 1

    def record_first_audio(self, latency_ms):
        self.first_audio_latencies.append(latency_ms)

Points clés à retenir

La documentation mentionne un permis de session de 600 secondes : comptez-le par prudence depuis la connexion
Implémentez la reconnexion automatique avec backoff exponentiel
Renouvelez la session avant ce terme (marge de 60 secondes)
Utilisez un pool de sessions pour les applications à fort trafic
Respectez la limite de 50 sessions concurrentes par équipe
Surveillez les métriques de connexion en production