TTS with API

curl -X POST "https://client.camb.ai/apis/tts-stream" \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "text": "Hello from the command line!",
    "voice_id": 147320,
    "language": "en-us",
    "speech_model": "mars-8.1-flash-beta",
    "output_configuration": {
      "format": "wav"
    }
  }' \
  --output output.wav

import os
import requests

def text_to_speech(text: str, voice_id: int = 147320) -> bytes:
    """Synchronous TTS call: stream the response and concatenate chunks (matches the streaming endpoint)."""
    api_key = os.getenv("CAMB_API_KEY")
    url = "https://client.camb.ai/apis/tts-stream"

    headers = {
        "x-api-key": api_key,
        "Content-Type": "application/json",
    }

    payload = {
        "text": text,
        "voice_id": voice_id,
        "language": "en-us",
        "speech_model": "mars-8.1-flash-beta",
        "output_configuration": {"format": "wav"},
    }

    with requests.post(url, headers=headers, json=payload, stream=True) as response:
        response.raise_for_status()
        return b"".join(
            chunk for chunk in response.iter_content(chunk_size=1024) if chunk
        )


if __name__ == "__main__":
    audio = text_to_speech("Hello world!")
    with open("output.wav", "wb") as f:
        f.write(audio)

import asyncio
import os
import aiohttp

async def text_to_speech(text: str, voice_id: int = 147320) -> bytes:
    """Convert text to speech using direct API call; buffers full body after a successful response."""
    api_key = os.getenv("CAMB_API_KEY")
    url = "https://client.camb.ai/apis/tts-stream"

    headers = {
        "x-api-key": api_key,
        "Content-Type": "application/json",
    }

    payload = {
        "text": text,
        "voice_id": voice_id,
        "language": "en-us",
        "speech_model": "mars-8.1-flash-beta",
        "output_configuration": {
            "format": "wav"
        },
    }

    timeout = aiohttp.ClientTimeout(total=120)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.post(url, headers=headers, json=payload) as resp:
            resp.raise_for_status()
            return await resp.read()


async def main():
    audio_data = await text_to_speech("Hello from the direct API!")

    with open("output.wav", "wb") as f:
        f.write(audio_data)

    print(f"Saved {len(audio_data)} bytes to output.wav")


if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import os
import aiohttp


async def list_voices():
    """List all available voices."""
    api_key = os.getenv("CAMB_API_KEY")
    url = "https://client.camb.ai/apis/list-voices"

    headers = {"x-api-key": api_key}

    async with aiohttp.ClientSession() as session:
        async with session.get(url, headers=headers) as resp:
            if resp.status == 200:
                voices = await resp.json()
                return voices
            else:
                raise Exception(f"Error: {resp.status}")


async def main():
    voices = await list_voices()

    print(f"Found {len(voices)} voices:\n")
    for voice in voices[:10]:  # Print first 10
        print(f"ID: {voice['id']}, Name: {voice['voice_name']}, Gender: {voice['gender']}")


if __name__ == "__main__":
    asyncio.run(main())

curl -X GET "https://client.camb.ai/apis/list-voices" \
  -H "x-api-key: YOUR_API_KEY"

import asyncio
import io
import os
import wave
import aiohttp
import numpy as np
import sounddevice as sd


async def play_tts(text: str):
    """Generate and play TTS audio."""
    api_key = os.getenv("CAMB_API_KEY")
    url = "https://client.camb.ai/apis/tts-stream"

    headers = {
        "x-api-key": api_key,
        "Content-Type": "application/json",
    }

    payload = {
        "text": text,
        "voice_id": 147320,
        "language": "en-us",
        "speech_model": "mars-8.1-flash-beta",
        "output_configuration": {"format": "wav"},
    }

    timeout = aiohttp.ClientTimeout(total=120)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.post(url, headers=headers, json=payload) as resp:
            resp.raise_for_status()
            audio_bytes = await resp.read()

    # Parse WAV and extract PCM data
    with wave.open(io.BytesIO(audio_bytes), 'rb') as wav_file:
        sample_rate = wav_file.getframerate()
        audio_data = np.frombuffer(wav_file.readframes(-1), dtype=np.int16)

    sd.play(audio_data, samplerate=sample_rate)
    sd.wait()


if __name__ == "__main__":
    asyncio.run(play_tts("Hello! This audio is playing directly."))

import wave

def save_as_wav(pcm_data: bytes, filename: str, sample_rate: int = 22050):
    """Save raw PCM16 mono bytes as a WAV file (use only when `output_configuration.format` is a raw `pcm_*` format)."""
    with wave.open(filename, "wb") as wav_file:
        wav_file.setnchannels(1)  # Mono
        wav_file.setsampwidth(2)  # 16-bit
        wav_file.setframerate(sample_rate)
        wav_file.writeframes(pcm_data)

# Example: wrap PCM from a `pcm_*` stream after you have the full byte buffer
# (If you use format "wav", the API already returns a WAV container—open it with wave.open(io.BytesIO(...)) instead.)
if __name__ == "__main__":
    pcm_bytes = b"..."  # your raw PCM payload
    save_as_wav(pcm_bytes, "output.wav", sample_rate=22050)