🔌 TTS Integrations

{
  "event": "tts.completed",
  "timestamp": "2025-11-27T10:30:00Z",
  "data": {
    "request_id": "req_abc123",
    "text": "Hello, this is a test.",
    "voice": "Peter",
    "model": "aura-prime",
    "language": "en-US",
    "characters": 23,
    "audio_url": "https://yourvoic.com/api/audio/abc123.mp3",
    "duration_ms": 1500
  }
}

import hmac
import hashlib

def verify_webhook(payload: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

{
  "text": "Hello, this is a test!",
  "voice": "Peter",
  "model": "aura-prime",
  "output_format": "mp3"
}

{
  "text": "Hello, this is a test!",
  "voice": "Peter",
  "model": "aura-prime",
  "output_format": "mp3",
  "response_format": "json"
}

{
  "success": true,
  "request_id": "abc123-def456",
  "audio": {
    "data": "base64-encoded-audio...",
    "format": "mp3",
    "mime_type": "audio/mpeg",
    "size_bytes": 12345,
    "data_uri": "data:audio/mpeg;base64,..."
  },
  "metadata": {
    "text": "Hello, this is a test!",
    "voice": "Peter",
    "model": "aura-prime",
    "characters_used": 24,
    "credits_used": 5,
    "processing_time_ms": 150
  }
}

{
  "text": "Hello, this is a streaming test!",
  "voice": "Peter",
  "model": "aura-prime"
}

{
  "url": "https://yourvoic.com/api/v1/tts/generate",
  "method": "POST",
  "headers": [
    { "name": "X-API-Key", "value": "your-api-key" },
    { "name": "Content-Type", "value": "application/json" }
  ],
  "body": {
    "text": "Your text here",
    "voice": "Peter",
    "model": "aura-prime",
    "language": "en-US"
  }
}

import httpx
from pipecat.services.tts import TTSService

class YourVoicTTS(TTSService):
    def __init__(self, api_key: str, voice: str = "Peter"):
        self.api_key = api_key
        self.voice = voice
        self.base_url = "https://yourvoic.com/api/v1"
    
    async def synthesize(self, text: str) -> bytes:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/tts/generate",
                headers={"X-API-Key": self.api_key},
                json={
                    "text": text,
                    "voice": self.voice,
                    "model": "aura-prime",
                    "output_format": "mp3"
                }
            )
            return response.content

# Usage in pipeline
tts = YourVoicTTS(api_key="your-key", voice="Peter")

from livekit.agents import tts
import httpx

class YourVoicTTSPlugin(tts.TTS):
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    async def synthesize(self, text: str) -> tts.SynthesizedAudio:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://yourvoic.com/api/v1/tts/generate",
                headers={"X-API-Key": self.api_key},
                json={"text": text, "voice": "Peter"}
            )
            return tts.SynthesizedAudio(
                audio=response.content,
                sample_rate=24000
            )

#!/usr/bin/env python3
"""YourVoic TTS AGI Script for Asterisk"""
import sys
import os
import requests
import hashlib
import tempfile

# Configuration
API_KEY = "your-api-key"
API_URL = "https://yourvoic.com/api/v1/tts/generate"
CACHE_DIR = "/var/lib/asterisk/sounds/tts-cache"

def agi_command(cmd):
    """Send AGI command and get response"""
    sys.stdout.write(f"{cmd}\n")
    sys.stdout.flush()
    return sys.stdin.readline().strip()

def generate_tts(text, voice="Peter", language="en-US"):
    """Generate TTS audio and cache it"""
    # Create cache filename based on text hash
    text_hash = hashlib.md5(f"{text}{voice}{language}".encode()).hexdigest()
    cache_file = os.path.join(CACHE_DIR, f"{text_hash}")
    wav_file = f"{cache_file}.wav"
    
    # Return cached file if exists
    if os.path.exists(wav_file):
        return cache_file
    
    # Generate new audio
    response = requests.post(API_URL, 
        headers={"X-API-Key": API_KEY},
        json={
            "text": text,
            "voice": voice,
            "model": "aura-prime",
            "language": language,
            "output_format": "wav",
            "sample_rate": 8000  # Telephony standard
        }
    )
    
    if response.status_code == 200:
        os.makedirs(CACHE_DIR, exist_ok=True)
        with open(wav_file, 'wb') as f:
            f.write(response.content)
        return cache_file
    return None

def main():
    # Read AGI environment
    env = {}
    while True:
        line = sys.stdin.readline().strip()
        if not line:
            break
        if ':' in line:
            key, value = line.split(':', 1)
            env[key.strip()] = value.strip()
    
    # Get text from AGI argument
    text = env.get('agi_arg_1', 'Hello, welcome to YourVoic.')
    voice = env.get('agi_arg_2', 'Peter')
    language = env.get('agi_arg_3', 'en-US')
    
    # Generate TTS
    audio_file = generate_tts(text, voice, language)
    
    if audio_file:
        # Play the audio
        agi_command(f"STREAM FILE {audio_file} #")
    else:
        agi_command("VERBOSE \"TTS generation failed\" 1")

if __name__ == "__main__":
    main()

[ivr-menu]
; Simple IVR with dynamic TTS
exten => s,1,Answer()
same => n,AGI(yourvoic_tts.py,"Welcome to our company. Press 1 for sales, 2 for support.","Peter","en-US")
same => n,WaitExten(5)

exten => 1,1,AGI(yourvoic_tts.py,"Connecting you to sales.","Peter","en-US")
same => n,Dial(SIP/sales)

exten => 2,1,AGI(yourvoic_tts.py,"Connecting you to support.","Peter","en-US")
same => n,Dial(SIP/support)

; Dynamic announcement
[announcements]
exten => _X.,1,Answer()
same => n,Set(MESSAGE=${ODBC_GET_ANNOUNCEMENT(${EXTEN})})
same => n,AGI(yourvoic_tts.py,"${MESSAGE}","Peter","en-US")
same => n,Hangup()

import ari
import requests
import tempfile
import os

# ARI connection
client = ari.connect('http://localhost:8088', 'asterisk', 'asterisk')

YOURVOIC_API_KEY = "your-api-key"
SOUNDS_DIR = "/var/lib/asterisk/sounds"

def generate_and_play_tts(channel, text, voice="Peter"):
    """Generate TTS and play to channel"""
    # Generate audio via YourVoic API
    response = requests.post(
        "https://yourvoic.com/api/v1/tts/generate",
        headers={"X-API-Key": YOURVOIC_API_KEY},
        json={
            "text": text,
            "voice": voice,
            "model": "aura-prime",
            "output_format": "wav",
            "sample_rate": 8000
        }
    )
    
    if response.status_code == 200:
        # Save to Asterisk sounds directory
        filename = f"tts_{channel.id}"
        filepath = os.path.join(SOUNDS_DIR, f"{filename}.wav")
        
        with open(filepath, 'wb') as f:
            f.write(response.content)
        
        # Play using ARI
        playback = channel.play(media=f'sound:{filename}')
        playback.on_event('PlaybackFinished', lambda p, e: cleanup(filepath))
        return playback
    return None

def cleanup(filepath):
    """Remove temporary audio file"""
    try:
        os.remove(filepath)
    except:
        pass

def stasis_start_cb(channel, ev):
    """Handle incoming channel"""
    channel.answer()
    generate_and_play_tts(channel, "Welcome to our IVR powered by YourVoic.")

client.on_channel_event('StasisStart', stasis_start_cb)
client.run(apps='yourvoic-tts')

-- YourVoic TTS for FreeSWITCH
local api_key = "your-api-key"
local api_url = "https://yourvoic.com/api/v1/tts/generate"
local cache_dir = "/var/lib/freeswitch/sounds/tts-cache/"

function yourvoic_tts(session, text, voice, language)
    voice = voice or "Peter"
    language = language or "en-US"
    
    -- Create cache filename
    local hash = session:md5(text .. voice .. language)
    local cache_file = cache_dir .. hash .. ".wav"
    
    -- Check cache
    local f = io.open(cache_file, "r")
    if f then
        f:close()
    else
        -- Generate via API
        local cmd = string.format(
            'curl -s -X POST "%s" -H "X-API-Key: %s" -H "Content-Type: application/json" ' ..
            '-d \'{"text":"%s","voice":"%s","language":"%s","model":"aura-prime",' ..
            '"output_format":"wav","sample_rate":8000}\' -o "%s"',
            api_url, api_key, text, voice, language, cache_file
        )
        os.execute(cmd)
    end
    
    -- Play audio
    session:streamFile(cache_file)
end

-- Usage in dialplan XML:
-- <action application="lua" data="yourvoic_tts.lua 'Hello world' Peter en-US"/>