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"
    }
    
    async with websockets.connect(url, additional_headers=headers) as ws:
        # First message: Init message (sets up session)
        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",  # Optional: Language code (ISO 639-1, defaults to 'en')
            "model": "voiceai-tts-v1-latest",  # Optional: model
            "flush": True  # Generate immediately
        }))
        
        # Receive audio chunks (JSON with base64-encoded audio)
        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 of audio")
                # Note: is_last is sent as a separate message after all chunks
                continue
            elif data.get("is_last"):
                # Completion message (sent after all audio chunks)
                print("Complete!")
                break
            elif data.get("error"):
                print(f"Error: {data['error']}")
                break
        
        # Save audio
        with open("output.mp3", "wb") as f:
            f.write(audio_data)
        
        # Subsequent messages: Text-only (voice_id, model, language are remembered)
        await ws.send(json.dumps({
            "text": "This is a second message.",
            "flush": True
        }))

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"):
                # Completion message (sent after all audio chunks)
                completed.add(ctx)
                print(f"{ctx} complete!")
        
        # Send text-only message to existing context
        await ws.send(json.dumps({
            "context_id": "ctx-1",
            "text": "More text for context one.",
            "flush": True
        }))
        
        # Close a specific context
        await ws.send(json.dumps({
            "context_id": "ctx-1",
            "close_context": True
        }))
        
        # 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())