import asyncio
import json
import base64
import websockets
async def tts_stream():
url = "wss://api.telnyx.com/v2/text-to-speech/speech?voice=FishAudio.s2.1-pro.933563129e564b19a115bedd57b7406a"
headers = {"Authorization": "Bearer YOUR_TELNYX_API_KEY"}
async with websockets.connect(url, additional_headers=headers) as ws:
# 1. Handshake
await ws.send(json.dumps({
"text": " ",
"voice_settings": {"format": "mp3", "sample_rate": 44100}
}))
# 2. Send text
await ws.send(json.dumps({"text": "Hello from Fish Audio on Telnyx."}))
# 3. Signal end of input
await ws.send(json.dumps({"text": ""}))
# 4. Collect audio
audio_chunks = []
async for message in ws:
data = json.loads(message)
if data.get("error"):
print(f"Error: {data['error']}")
break
if data.get("audio"):
audio_chunks.append(base64.b64decode(data["audio"]))
if data.get("isFinal"):
break
with open("output.mp3", "wb") as f:
for chunk in audio_chunks:
f.write(chunk)
asyncio.run(tts_stream())