TTS via RAW API Calls

Overview

While SDKs and frameworks provide convenience, sometimes you need direct control over API calls. This tutorial shows how to call the Camb.ai TTS API directly using HTTP requests.

When to Use Direct API

Building integrations in languages without an SDK
Need fine-grained control over request/response handling
Debugging or testing API behavior
Building custom streaming implementations

Listen to an Example

Prerequisites

Create an account

Get your API key

Go to Settings → API Keys in Studio and copy your key. See Authentication for details.

Install the SDK

pip install camb-ai

Skip this step if you’re using the direct API.

Set your API key to use in your code

export CAMB_API_KEY="your_api_key_here"

Basic TTS Request

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-flash",
    "output_configuration": {
      "format": "wav"
    }
  }' \
  --output output.wav

Streaming Response

For real-time applications, stream audio chunks as they’re generated:

import asyncio
import os
import time
import aiohttp


async def stream_tts(text: str, voice_id: int = 147320):
    """Stream TTS audio with timing metrics."""
    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-flash",
        "output_configuration": {"format": "wav"},
    }

    async with aiohttp.ClientSession() as session:
        start_time = time.perf_counter()
        ttfb = None

        async with session.post(url, headers=headers, json=payload) as resp:
            print(f"Status: {resp.status}")
            print(f"Content-Type: {resp.headers.get('Content-Type')}")

            chunk_count = 0
            total_bytes = 0

            async for chunk in resp.content.iter_chunked(4096):
                if ttfb is None:
                    ttfb = (time.perf_counter() - start_time) * 1000
                    print(f"Time to First Byte: {ttfb:.0f}ms")

                chunk_count += 1
                total_bytes += len(chunk)
                yield chunk

            total_time = (time.perf_counter() - start_time) * 1000
            print(f"\nTotal: {chunk_count} chunks, {total_bytes} bytes")
            print(f"TTFB: {ttfb:.0f}ms, Total: {total_time:.0f}ms")


async def main():
    with open("streamed_output.wav", "wb") as f:
        async for chunk in stream_tts("This is streamed audio generation."):
            f.write(chunk)


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

Request Parameters

Required Parameters

Parameter	Type	Description
`text`	string	Text to convert to speech (min 3 characters)
`voice_id`	integer	Voice ID to use

Optional Parameters

Parameter	Type	Default	Description
`language`	string	`"en-us"`	BCP-47 language code
`speech_model`	string	`"mars-flash"`	Model: `mars-flash`, `mars-pro`, `mars-instruct` (`mars-instruct` supports expressive text cues)
`output_configuration`	object	`{}`	Output format settings
`enhance_named_entities`	boolean	`false`	Better pronunciation for names/places

More details available in the API Reference.

Expressive Input with `mars-instruct`

When you use speech_model: "mars-instruct", you can encode expression directly in the text field by adding short emotion or pacing cues. English examples:

[speaking slowly] This is very important. Please pay close attention.
[excited] We shipped the feature, and the response has been fantastic!
Let's pause for a moment <break time="400ms"/> and continue clearly.

This is useful when you want fine control over delivery without relying only on user_instructions.

For a comprehensive guide on emotional expression, pauses, and prosody control, see the Emotional Voice Control tutorial.

Listing Voices

Get available voices:

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())

Error Handling

Handle common errors gracefully:

import asyncio
import os
import aiohttp


class CambAPIError(Exception):
    def __init__(self, status: int, message: str):
        self.status = status
        self.message = message
        super().__init__(f"API Error {status}: {message}")


async def text_to_speech_safe(text: str, voice_id: int = 147320):
    """TTS with proper error handling."""
    api_key = os.getenv("CAMB_API_KEY")

    if not api_key:
        raise ValueError("CAMB_API_KEY environment variable not set")

    if len(text) < 3:
        raise ValueError("Text must be at least 3 characters")

    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-flash",
        "output_configuration": {"format": "wav"},
    }

    timeout = aiohttp.ClientTimeout(total=60)

    async with aiohttp.ClientSession(timeout=timeout) as session:
        try:
            async with session.post(url, headers=headers, json=payload) as resp:
                if resp.status == 200:
                    return await resp.read()
                elif resp.status == 401:
                    raise CambAPIError(401, "Invalid API key")
                elif resp.status == 400:
                    error = await resp.json()
                    raise CambAPIError(400, error.get("message", "Bad request"))
                elif resp.status == 429:
                    raise CambAPIError(429, "Rate limit exceeded")
                else:
                    raise CambAPIError(resp.status, await resp.text())

        except asyncio.TimeoutError:
            raise CambAPIError(408, "Request timed out")
        except aiohttp.ClientError as e:
            raise CambAPIError(500, f"Connection error: {str(e)}")


async def main():
    try:
        audio = await text_to_speech_safe("Hello world!")
        with open("output.wav", "wb") as f:
            f.write(audio)
        print(f"Saved {len(audio)} bytes to output.wav")
    except ValueError as e:
        print(f"Validation error: {e}")
    except CambAPIError as e:
        print(f"API error: {e}")


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

Playing Audio

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-flash",
        "output_configuration": {"format": "wav"},
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(url, headers=headers, json=payload) as resp:
            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."))

Next Steps

Python SDK

Use the SDK for simpler integration

API Reference

Complete API documentation

https://mintcdn.com/cambai/2LvnefIkletroPxv/images/pipecat-orange.svg?fit=max&auto=format&n=2LvnefIkletroPxv&q=85&s=40cf8e001b8cadc8a4c3c557dea603d5

Voice Agents

Build real-time voice applications

Voice Library

Browse available voices

Getting Started

Models

SDK Guides

Tutorials

Hosting Platforms

Integrations

API Reference

TTS via RAW API Calls

Overview

When to Use Direct API

Listen to an Example

Prerequisites

Basic TTS Request

Streaming Response

Request Parameters

Required Parameters

Optional Parameters

Expressive Input with `mars-instruct`

Listing Voices

Error Handling

Playing Audio

Next Steps

Python SDK

API Reference

Voice Agents

Voice Library

Getting Started

Models

SDK Guides

Tutorials

Hosting Platforms

Integrations

API Reference

​Overview

​When to Use Direct API

​Listen to an Example

​Prerequisites

​Basic TTS Request

​Streaming Response

​Request Parameters

​Required Parameters

​Optional Parameters

​Expressive Input with mars-instruct

​Listing Voices

​Error Handling

​Playing Audio

​Next Steps

Python SDK

API Reference

Voice Agents

Voice Library

Overview

When to Use Direct API

Listen to an Example

Prerequisites

Basic TTS Request

Streaming Response

Request Parameters

Required Parameters

Optional Parameters

Expressive Input with `mars-instruct`

Listing Voices

Error Handling

Playing Audio

Next Steps