Cartesia

pip install cartesia netra-sdk

npm install @cartesia/cartesia-js netra-sdk

import os
from netra import Netra

Netra.init(
    app_name="cartesia-service",
    headers=f"x-api-key={os.environ.get('NETRA_API_KEY')}"
)

import Netra from 'netra-sdk';

await Netra.init({
  appName: 'cartesia-service',
  headers: `x-api-key=${process.env.NETRA_API_KEY}`
});

from cartesia import Cartesia
from netra.decorators import task, workflow
import os

client = Cartesia(api_key=os.environ.get("CARTESIA_API_KEY"))

@task()
def generate_speech(text: str, voice_id: str) -> bytes:
    """Generate speech using Cartesia Sonic model."""
    response = client.tts.bytes(
        model_id="sonic-english",
        transcript=text,
        voice={
            "mode": "id",
            "id": voice_id
        },
        output_format={
            "container": "raw",
            "encoding": "pcm_f32le",
            "sample_rate": 44100
        }
    )
    
    return response

@task()
def stream_speech(text: str, voice_id: str):
    """Stream speech using Cartesia Sonic Turbo."""
    stream = client.tts.stream(
        model_id="sonic-turbo",
        transcript=text,
        voice={
            "mode": "id",
            "id": voice_id
        },
        output_format={
            "container": "raw",
            "encoding": "pcm_f32le",
            "sample_rate": 44100
        }
    )
    
    for chunk in stream:
        yield chunk

@workflow()
def process_text_batch(texts: list[str], voice_id: str) -> list[bytes]:
    """Process multiple texts to speech."""
    audio_buffers = []
    
    for text in texts:
        audio = generate_speech(text, voice_id)
        audio_buffers.append(audio)
    
    return audio_buffers

# Usage
audio_data = generate_speech(
    "Hello, this is Cartesia Sonic speech synthesis.",
    "a0e99841-438c-4a64-b679-ae501e7d6091"
)

import Cartesia from '@cartesia/cartesia-js';
import { task, workflow } from 'netra-sdk';

const cartesia = new Cartesia({
  apiKey: process.env.CARTESIA_API_KEY
});

class TTSService {
  @task()
  async generateSpeech(text: string, voiceId: string): Promise<Buffer> {
    const response = await cartesia.tts.bytes({
      model_id: 'sonic-english',
      transcript: text,
      voice: {
        mode: 'id',
        id: voiceId
      },
      output_format: {
        container: 'raw',
        encoding: 'pcm_f32le',
        sample_rate: 44100
      }
    });

    return Buffer.from(response);
  }

  @task()
  async streamSpeech(text: string, voiceId: string): Promise<AsyncGenerator<Buffer>> {
    const stream = await cartesia.tts.stream({
      model_id: 'sonic-turbo',
      transcript: text,
      voice: {
        mode: 'id',
        id: voiceId
      },
      output_format: {
        container: 'raw',
        encoding: 'pcm_f32le',
        sample_rate: 44100
      }
    });

    async function* generateChunks() {
      for await (const chunk of stream) {
        yield Buffer.from(chunk);
      }
    }

    return generateChunks();
  }

  @workflow()
  async processTextBatch(texts: string[], voiceId: string): Promise<Buffer[]> {
    const audioBuffers: Buffer[] = [];
    
    for (const text of texts) {
      const audio = await this.generateSpeech(text, voiceId);
      audioBuffers.push(audio);
    }
    
    return audioBuffers;
  }
}

// Usage
const service = new TTSService();
const audio = await service.generateSpeech(
  'Hello, this is Cartesia Sonic speech synthesis.',
  'a0e99841-438c-4a64-b679-ae501e7d6091'
);

from cartesia import Cartesia
from netra.decorators import task
import os

client = Cartesia(api_key=os.environ.get("CARTESIA_API_KEY"))

@task()
def transcribe_audio(audio_path: str) -> str:
    """Transcribe audio using Cartesia Ink model."""
    with open(audio_path, "rb") as audio_file:
        audio_data = audio_file.read()
    
    response = client.stt.transcribe(
        model_id="ink-whisper",
        audio=audio_data,
        language="en"
    )
    
    return response.transcript

@task()
def transcribe_stream(audio_stream) -> str:
    """Transcribe streaming audio in real-time."""
    full_transcript = ""
    
    stream = client.stt.stream_transcribe(
        model_id="ink-whisper",
        audio=audio_stream,
        language="en",
        interim_results=True
    )
    
    for result in stream:
        if result.is_final:
            full_transcript += result.transcript + " "
    
    return full_transcript.strip()

# Usage
transcript = transcribe_audio("./audio/sample.wav")

import Cartesia from '@cartesia/cartesia-js';
import { task } from 'netra-sdk';
import * as fs from 'fs';

const cartesia = new Cartesia({
  apiKey: process.env.CARTESIA_API_KEY
});

class STTService {
  @task()
  async transcribeAudio(audioPath: string): Promise<string> {
    const audioBuffer = fs.readFileSync(audioPath);
    
    const response = await cartesia.stt.transcribe({
      model_id: 'ink-whisper',
      audio: audioBuffer,
      language: 'en'
    });

    return response.transcript;
  }

  @task()
  async transcribeStream(audioStream: ReadableStream): Promise<string> {
    let fullTranscript = '';
    
    const stream = await cartesia.stt.streamTranscribe({
      model_id: 'ink-whisper',
      audio: audioStream,
      language: 'en',
      interim_results: true
    });

    for await (const result of stream) {
      if (result.is_final) {
        fullTranscript += result.transcript + ' ';
      }
    }

    return fullTranscript.trim();
  }
}

// Usage
const sttService = new STTService();
const transcript = await sttService.transcribeAudio('./audio/sample.wav');

from cartesia import Cartesia
from netra import SpanWrapper, ActionModel, UsageModel
import os
import time

client = Cartesia(api_key=os.environ.get("CARTESIA_API_KEY"))

def generate_speech_with_tracking(text: str, voice_id: str) -> bytes:
    """Generate speech with detailed tracking."""
    span = SpanWrapper("cartesia-tts")
    span.start()
    
    try:
        start_time = time.time_ns()
        span.set_attribute("text_length", len(text))
        span.set_attribute("voice_id", voice_id)
        span.set_attribute("model", "sonic-turbo")
        
        response = client.tts.bytes(
            model_id="sonic-turbo",
            transcript=text,
            voice={"mode": "id", "id": voice_id},
            output_format={
                "container": "raw",
                "encoding": "pcm_f32le",
                "sample_rate": 44100
            }
        )
        
        end_time = time.time_ns()
        duration_ms = (end_time - start_time) / 1_000_000
        
        # Track the TTS API operation
        action = ActionModel(
            start_time=str(start_time),
            action="API",
            action_type="TTS_SYNTHESIS",
            metadata={
                "provider": "cartesia",
                "model": "sonic-turbo",
                "voice_id": voice_id,
                "text_length": str(len(text)),
                "audio_size_bytes": str(len(response)),
                "sample_rate": "44100",
                "latency_ms": str(duration_ms)
            },
            success=True
        )
        span.set_action([action])
        
        # Track usage
        usage = UsageModel(
            model="sonic-turbo",
            usage_type="characters",
            units_used=len(text),
            cost_in_usd=len(text) * 0.00001
        )
        span.set_usage([usage])
        
        span.set_status({"code": 1, "message": "Success"})
        span.end()
        
        return response
        
    except Exception as e:
        span.set_error(e)
        span.set_status({"code": 2, "message": "Error"})
        span.end()
        raise

def transcribe_with_tracking(audio_path: str) -> str:
    """Transcribe audio with detailed tracking."""
    span = SpanWrapper("cartesia-stt")
    span.start()
    
    try:
        start_time = time.time_ns()
        
        with open(audio_path, "rb") as audio_file:
            audio_data = audio_file.read()
        
        audio_size_bytes = len(audio_data)
        span.set_attribute("audio_file", audio_path)
        span.set_attribute("audio_size_bytes", audio_size_bytes)
        span.set_attribute("model", "ink-whisper")
        
        response = client.stt.transcribe(
            model_id="ink-whisper",
            audio=audio_data,
            language="en"
        )
        
        end_time = time.time_ns()
        duration_ms = (end_time - start_time) / 1_000_000
        
        # Track the STT API operation
        action = ActionModel(
            start_time=str(start_time),
            action="API",
            action_type="STT_TRANSCRIPTION",
            metadata={
                "provider": "cartesia",
                "model": "ink-whisper",
                "audio_size_bytes": str(audio_size_bytes),
                "transcript_length": str(len(response.transcript)),
                "duration_ms": str(duration_ms),
                "language": "en"
            },
            success=True
        )
        span.set_action([action])
        
        # Track usage
        audio_duration = getattr(response, 'duration', 0)
        usage = UsageModel(
            model="ink-whisper",
            usage_type="audio_seconds",
            units_used=audio_duration,
            cost_in_usd=audio_duration * 0.036  # $0.13 per hour
        )
        span.set_usage([usage])
        
        span.set_attribute("transcript_length", len(response.transcript))
        span.set_status({"code": 1, "message": "Success"})
        span.end()
        
        return response.transcript
        
    except Exception as e:
        span.set_error(e)
        span.set_status({"code": 2, "message": "Error"})
        span.end()
        raise

# Usage
audio_data = generate_speech_with_tracking(
    "This is ultra-low latency speech synthesis.",
    "a0e99841-438c-4a64-b679-ae501e7d6091"
)

transcript = transcribe_with_tracking("./audio/sample.wav")

import Cartesia from '@cartesia/cartesia-js';
import { SpanWrapper, ActionModel } from 'netra-sdk';

const cartesia = new Cartesia({
  apiKey: process.env.CARTESIA_API_KEY
});

async function generateSpeechWithTracking(text: string, voiceId: string): Promise<Buffer> {
  const span = new SpanWrapper('cartesia-tts');
  span.start();

  try {
    const startTime = Date.now();
    span.setAttribute('text_length', text.length);
    span.setAttribute('voice_id', voiceId);
    span.setAttribute('model', 'sonic-turbo');

    const response = await cartesia.tts.bytes({
      model_id: 'sonic-turbo',
      transcript: text,
      voice: { mode: 'id', id: voiceId },
      output_format: {
        container: 'raw',
        encoding: 'pcm_f32le',
        sample_rate: 44100
      }
    });

    const audioBuffer = Buffer.from(response);
    const duration = Date.now() - startTime;

    // Track the TTS API operation
    const action: ActionModel = {
      start_time: (startTime * 1000000).toString(),
      action: 'API',
      action_type: 'TTS_SYNTHESIS',
      metadata: {
        provider: 'cartesia',
        model: 'sonic-turbo',
        voice_id: voiceId,
        text_length: text.length.toString(),
        audio_size_bytes: audioBuffer.length.toString(),
        sample_rate: '44100',
        latency_ms: duration.toString()
      },
      success: true
    };
    span.setAction([action]);

    span.setUsage({
      model: 'sonic-turbo',
      usage_type: 'characters',
      units_used: text.length,
      cost_in_usd: text.length * 0.00001
    });

    span.setStatus({ code: 1, message: 'Success' });
    span.end();

    return audioBuffer;
  } catch (error) {
    span.setError(error as Error);
    span.setStatus({ code: 2, message: 'Error' });
    span.end();
    throw error;
  }
}

async function transcribeWithTracking(audioPath: string): Promise<string> {
  const span = new SpanWrapper('cartesia-stt');
  span.start();

  try {
    const startTime = Date.now();
    const audioBuffer = fs.readFileSync(audioPath);
    const audioSizeBytes = audioBuffer.length;

    span.setAttribute('audio_file', audioPath);
    span.setAttribute('audio_size_bytes', audioSizeBytes);
    span.setAttribute('model', 'ink-whisper');

    const response = await cartesia.stt.transcribe({
      model_id: 'ink-whisper',
      audio: audioBuffer,
      language: 'en'
    });

    const duration = Date.now() - startTime;

    // Track the STT API operation
    const action: ActionModel = {
      start_time: (startTime * 1000000).toString(),
      action: 'API',
      action_type: 'STT_TRANSCRIPTION',
      metadata: {
        provider: 'cartesia',
        model: 'ink-whisper',
        audio_size_bytes: audioSizeBytes.toString(),
        transcript_length: response.transcript.length.toString(),
        duration_ms: duration.toString(),
        language: 'en'
      },
      success: true
    };
    span.setAction([action]);

    span.setUsage({
      model: 'ink-whisper',
      usage_type: 'audio_seconds',
      units_used: response.duration || 0,
      cost_in_usd: (response.duration || 0) * 0.036
    });

    span.setAttribute('transcript_length', response.transcript.length);
    span.setStatus({ code: 1, message: 'Success' });
    span.end();

    return response.transcript;
  } catch (error) {
    span.setError(error as Error);
    span.setStatus({ code: 2, message: 'Error' });
    span.end();
    throw error;
  }
}

// Usage
const audio = await generateSpeechWithTracking(
  'This is ultra-low latency speech synthesis.',
  'a0e99841-438c-4a64-b679-ae501e7d6091'
);

const transcript = await transcribeWithTracking('./audio/sample.wav');