Text-to-Speech Streaming API

import requests

# Using default voice (voice_id is optional)
response = requests.post(
    'https://dev.voice.ai/api/v1/tts/speech/stream',
    headers={'Authorization': 'Bearer YOUR_API_KEY', 'Content-Type': 'application/json'},
    json={
        'text': 'This is a test of streaming audio generation.',
        'model': 'voiceai-tts-v1-latest',  # Optional, defaults to voiceai-tts-v1-latest
        'language': 'en'  # Optional, defaults to 'en'
    },
    stream=True
)

# Or with a custom voice_id:
# json={'voice_id': 'your-voice-id-here', 'text': 'This is a test of streaming audio generation.', 'model': 'voiceai-tts-v1-latest', 'language': 'en'}

with open('output.mp3', 'wb') as f:
    for chunk in response.iter_content():
        if chunk: f.write(chunk)

import asyncio
import json
import base64
import websockets

async def test_tts_websocket():
    url = "wss://dev.voice.ai/api/v1/tts/stream"
    headers = {
        "Authorization": "Bearer YOUR_API_KEY"
    }
    
    try:
        async with websockets.connect(url, additional_headers=headers) as ws:
            # Init message with text and flush=true for immediate generation
            await ws.send(json.dumps({
                "voice_id": None,  # Optional: voice ID (None = default built-in voice)
                "text": "Hello, this is a test.",
                "audio_format": "mp3",
                "language": "en",
                "model": "voiceai-tts-v1-latest",
                "flush": True
            }))
            
            # Receive audio chunks
            audio_data = b""
            while True:
                msg = await ws.recv()
                data = json.loads(msg)
                if data.get("audio"):
                    chunk = base64.b64decode(data["audio"])
                    audio_data += chunk
                    print(f"Received {len(chunk)} bytes")
                elif data.get("is_last"):
                    print("Complete!")
                    break
            
            # Save audio
            with open("output.mp3", "wb") as f:
                f.write(audio_data)
                
    except websockets.ConnectionClosed as e:
        # Errors close connection with close code (no JSON error message)
        print(f"Connection closed: code={e.code} reason={e.reason}")

asyncio.run(test_tts_websocket())

import asyncio
import json
import base64
import websockets

async def test_multi_context():
    url = "wss://dev.voice.ai/api/v1/tts/multi-stream"
    headers = {
        "Authorization": "Bearer YOUR_API_KEY",
    }
    
    async with websockets.connect(url, additional_headers=headers) as ws:
        # Init context-1 (first message to ctx-1)
        await ws.send(json.dumps({
            "context_id": "ctx-1",
            "voice_id": "VOICE_ID_1",
            "text": "Hello from context one.",
            "language": "en",
            "model": "voiceai-tts-v1-latest",
            "flush": True
        }))
        
        # Init context-2 (first message to ctx-2, can use different voice)
        await ws.send(json.dumps({
            "context_id": "ctx-2", 
            "voice_id": "VOICE_ID_2",
            "text": "Hello from context two.",
            "language": "en",
            "model": "voiceai-tts-v1-latest",
            "flush": True
        }))
        
        # Receive responses (will be interleaved, all JSON with base64 audio)
        audio_by_context = {}  # {context_id: bytes}
        completed = set()
        while len(completed) < 2:
            msg = await ws.recv()
            data = json.loads(msg)
            ctx = data.get("context_id")
            if data.get("audio"):
                chunk = base64.b64decode(data["audio"])
                if ctx not in audio_by_context:
                    audio_by_context[ctx] = b""
                audio_by_context[ctx] += chunk
                print(f"Audio chunk for {ctx}: {len(chunk)} bytes")
                # Note: is_last is sent as a separate message after all chunks
                continue
            elif data.get("is_last"):
                # Flush completion (sent after all audio chunks for this flush)
                # Context remains active and can receive more flushes
                completed.add(ctx)
                print(f"{ctx} flush complete!")
        
        # Send another message to existing context
        await ws.send(json.dumps({
            "context_id": "ctx-1",
            "text": "More text for context one.",
            "flush": True
        }))
        
        # Receive audio for the new message
        while True:
            msg = await ws.recv()
            data = json.loads(msg)
            if data.get("audio"):
                chunk = base64.b64decode(data["audio"])
                audio_by_context["ctx-1"] += chunk
            elif data.get("is_last"):
                break
        
        # Close entire socket
        await ws.send(json.dumps({"close_socket": True}))
    
    # Save audio to files
    for ctx_id, audio_data in audio_by_context.items():
        filename = f"output_{ctx_id}.mp3"
        with open(filename, "wb") as f:
            f.write(audio_data)
        print(f"Saved {ctx_id} audio to {filename} ({len(audio_data)} bytes)")

asyncio.run(test_multi_context())