Managing WebSocket sessions — Text-to-Speech API

Updated November 3, 2026, based on the xAI Text to Speech page, which mentions a 600-second “session permit” and a connection with no timeout. The renewal strategy below is a precaution. Source: Text to Speech.

Master the session life cycle

In production, robust WebSocket session management is what separates a prototype from a reliable product. TTS sessions come with specific constraints (lifetime, concurrency limits) that your app has to handle properly.

Session life cycle

A TTS WebSocket session goes through this cycle:

Opening: connect to wss://api.x.ai/v1/tts with the parameters
Active: exchange text.delta/audio.delta messages
Normal closing: the client or the server closes the connection
Unexpected closing: an error, maintenance, or the end of the 600-second session permit the documentation mentions

Detect closing

The server can close the connection at any time (internal error, maintenance, end of the session permit). Your app should always handle the close event:

async def managed_session(api_key, language="en", 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"Connection closed: code={e.code}, reason={e.reason}")
    except websockets.ConnectionClosedError:
        print("Connection lost unexpectedly")

Automatic reconnection

For long-running apps, implement reconnection with exponential backoff:

import asyncio
import random

class TTSSessionManager:
    def __init__(self, api_key, language="en", 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"Connected (attempt {attempt + 1})")
                return
            except Exception as e:
                wait = min(2 ** attempt + random.random(), 30)
                print(f"Connection failed: {e}. Retrying in {wait:.1f}s")
                await asyncio.sleep(wait)

        raise ConnectionError("Could not connect after 5 attempts")

    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:
            # Reconnect and retry
            await self.connect()
            return await self.synthesize(text)

Managing session lifetime

The xAI Text to Speech page mentions a 600-second “session permit” without saying when it starts or what happens when it ends, and separately states that the connection has no timeout. The most cautious approach is to count from the moment you connect and renew the session before that point:

import time

class TTLAwareSession:
    TTL = 600  # seconds
    REFRESH_MARGIN = 60  # reconnect 60 s before expiry

    def __init__(self):
        self.connected_at = None

    async def connect(self):
        # ... connection ...
        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()  # New session
        # ... send the text ...

Reconnecting 60 seconds before expiry keeps you from being cut off mid-synthesis.

Session pool

For high-traffic apps, keep a pool of reusable sessions:

class SessionPool:
    def __init__(self, api_key, pool_size=10):
        self.api_key = api_key
        self.pool_size = min(pool_size, 50)  # Max 50 per team
        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

        # Wait for a session to free up
        return await self.available.get()

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

Keep the pool size to a fraction of the 50-session limit to leave headroom for your app’s other services.

Metrics to watch

To keep sessions healthy in production, monitor:

Number of active sessions: must not exceed 50
Reconnection rate: a high rate points to a network or lifetime problem
Latency of the first audio.delta: measures how responsive the service is
WebSocket errors: abnormal close codes (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)

Key takeaways

The documentation mentions a 600-second session permit: to be safe, count it from the moment you connect
Implement automatic reconnection with exponential backoff
Renew the session before that point (with a 60-second margin)
Use a session pool for high-traffic apps
Stay within the limit of 50 concurrent sessions per team
Monitor connection metrics in production