Code examples by language

Language-specific patterns for integrating with the Modulate APIs. Each section covers the key patterns for that language: authentication setup, file upload, WebSocket connections, and error handling. For full per-endpoint request and response schemas, see the API Reference tab.

cURL

cURL works for all HTTP batch endpoints. WebSocket endpoints require a dedicated tool — see WebSocket testing below.

Batch file upload with optional features

curl -X POST https://platform.modulate.ai/api/velma-2-stt-batch \
  -H "X-API-Key: YOUR_API_KEY" \
  -F "upload_file=@/path/to/recording.mp3" \
  -F "speaker_diarization=true" \
  -F "emotion_signal=true" \
  -F "accent_signal=false" \
  -F "pii_phi_tagging=false"

Replace the endpoint and fields for other batch APIs. The X-API-Key header pattern is the same for all batch endpoints.

Deepfake Detection batch — minimal example

curl -X POST https://platform.modulate.ai/api/velma-2-synthetic-voice-detection-batch \
  -H "X-API-Key: YOUR_API_KEY" \
  -F "upload_file=@/path/to/audio.mp3"

Python

Setup

# Synchronous
pip install requests

# Async
pip install aiohttp

Authentication — synchronous (requests)

Pass the API key in the headers dict. Reuse the session or headers across requests.

import requests

HEADERS = {"X-API-Key": "YOUR_API_KEY"}

response = requests.post(
    "https://platform.modulate.ai/api/velma-2-stt-batch",
    headers=HEADERS,
    files={"upload_file": open("recording.mp3", "rb")},
    data={"speaker_diarization": "true"},
)
response.raise_for_status()
result = response.json()

Authentication — async (aiohttp)

import aiohttp

async with aiohttp.ClientSession(headers={"X-API-Key": "YOUR_API_KEY"}) as session:
    # reuse session for multiple requests
    pass

Batch file upload (aiohttp)

import asyncio
import aiohttp

async def transcribe(file_path: str) -> dict:
    data = aiohttp.FormData()
    data.add_field(
        "upload_file",
        open(file_path, "rb"),
        filename=file_path.split("/")[-1],
        content_type="application/octet-stream",
    )
    data.add_field("speaker_diarization", "true")
    data.add_field("emotion_signal", "true")

    async with aiohttp.ClientSession(headers={"X-API-Key": "YOUR_API_KEY"}) as session:
        async with session.post(
            "https://platform.modulate.ai/api/velma-2-stt-batch",
            data=data,
        ) as resp:
            resp.raise_for_status()
            return await resp.json()

result = asyncio.run(transcribe("recording.mp3"))

Concurrent batch processing

Use a semaphore to respect your organization’s concurrent request limit.

import asyncio
import glob
import aiohttp

URL = "https://platform.modulate.ai/api/velma-2-stt-batch-english-vfast"
MAX_CONCURRENT = 5  # set to your organization's limit

semaphore = asyncio.Semaphore(MAX_CONCURRENT)

async def process_file(session: aiohttp.ClientSession, filepath: str) -> dict:
    async with semaphore:
        data = aiohttp.FormData()
        data.add_field(
            "upload_file",
            open(filepath, "rb"),
            filename=filepath.rsplit("/", 1)[-1],
            content_type="application/octet-stream",
        )
        async with session.post(URL, data=data) as resp:
            if resp.status != 200:
                return {"file": filepath, "error": f"{resp.status}: {await resp.text()}"}
            result = await resp.json()
            return {"file": filepath, **result}

async def main():
    files = glob.glob("audio_files/*.opus")
    async with aiohttp.ClientSession(headers={"X-API-Key": "YOUR_API_KEY"}) as session:
        results = await asyncio.gather(*[process_file(session, f) for f in files])
    for r in results:
        status = "OK" if "error" not in r else f"FAILED: {r['error']}"
        print(f"{r['file']}: {status}")

asyncio.run(main())

WebSocket streaming (aiohttp)

Applies to both the Multilingual Transcription streaming and Deepfake Detection streaming endpoints.

import asyncio
import json
import aiohttp

API_KEY = "YOUR_API_KEY"
AUDIO_FILE = "recording.opus"
CHUNK_SIZE = 8192

async def stream_transcription():
    url = (
        f"wss://platform.modulate.ai/api/velma-2-stt-streaming"
        f"?api_key={API_KEY}"
        f"&speaker_diarization=true"
        f"&emotion_signal=true"
    )

    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(url) as ws:

            async def send_audio():
                with open(AUDIO_FILE, "rb") as f:
                    while chunk := f.read(CHUNK_SIZE):
                        await ws.send_bytes(chunk)
                        await asyncio.sleep(CHUNK_SIZE / 4000)
                await ws.send_str("")

            send_task = asyncio.create_task(send_audio())

            try:
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.TEXT:
                        data = json.loads(msg.data)
                        if data["type"] == "utterance":
                            u = data["utterance"]
                            print(f"[Speaker {u['speaker']}] {u['text']}")
                        elif data["type"] == "done":
                            print(f"Done. Duration: {data['duration_ms']}ms")
                            break
                        elif data["type"] == "error":
                            print(f"Error: {data['error']}")
                            break
                    elif msg.type in (
                        aiohttp.WSMsgType.ERROR,
                        aiohttp.WSMsgType.CLOSE,
                        aiohttp.WSMsgType.CLOSED,
                    ):
                        break
            finally:
                if not send_task.done():
                    send_task.cancel()

asyncio.run(stream_transcription())

Error handling

import requests
from requests.exceptions import HTTPError

try:
    response = requests.post(url, headers=headers, files=files)
    response.raise_for_status()
    result = response.json()
except HTTPError as e:
    status = e.response.status_code
    detail = e.response.json().get("detail", "unknown error")
    if status == 401:
        print("Invalid or missing API key")
    elif status == 403:
        print(f"Access denied: {detail}")
    elif status == 429:
        print("Rate limit exceeded — wait and retry")
    elif status >= 500:
        print(f"Server error ({status}): {detail} — retry after delay")

JavaScript (Node.js)

Setup

npm install ws form-data node-fetch
# or use built-in fetch (Node 18+) and the ws package for WebSocket

Batch file upload (native fetch, Node 18+)

import fs from "fs";
import path from "path";

const API_KEY = "YOUR_API_KEY";

async function transcribe(filePath) {
  const formData = new FormData();
  formData.append(
    "upload_file",
    new Blob([fs.readFileSync(filePath)]),
    path.basename(filePath)
  );
  formData.append("speaker_diarization", "true");
  formData.append("emotion_signal", "true");

  const response = await fetch(
    "https://platform.modulate.ai/api/velma-2-stt-batch",
    {
      method: "POST",
      headers: { "X-API-Key": API_KEY },
      body: formData,
    }
  );

  if (!response.ok) {
    const err = await response.json().catch(() => ({ detail: response.statusText }));
    throw new Error(`${response.status}: ${err.detail}`);
  }

  return response.json();
}

const result = await transcribe("recording.mp3");
console.log(result.text);
for (const u of result.utterances) {
  console.log(`[Speaker ${u.speaker}] (${u.language}) ${u.text}`);
}

Batch file upload (form-data package, Node < 18 or streams)

Use the form-data package when you want to stream the file rather than reading it fully into memory.

import fs from "fs";
import path from "path";
import FormData from "form-data";

const API_KEY = "YOUR_API_KEY";

async function transcribe(filePath) {
  const form = new FormData();
  form.append("upload_file", fs.createReadStream(filePath), {
    filename: path.basename(filePath),
  });
  form.append("speaker_diarization", "true");

  const response = await fetch(
    "https://platform.modulate.ai/api/velma-2-stt-batch",
    {
      method: "POST",
      headers: {
        "X-API-Key": API_KEY,
        ...form.getHeaders(),
      },
      body: form,
    }
  );

  if (!response.ok) throw new Error(`${response.status}: ${await response.text()}`);
  return response.json();
}

WebSocket streaming (ws package)

const WebSocket = require("ws");
const fs = require("fs");

const API_KEY = "YOUR_API_KEY";
const CHUNK_SIZE = 8192;

const url = new URL("wss://platform.modulate.ai/api/velma-2-stt-streaming");
url.searchParams.set("api_key", API_KEY);
url.searchParams.set("speaker_diarization", "true");
url.searchParams.set("emotion_signal", "true");

const ws = new WebSocket(url.toString());

ws.on("open", () => {
  const stream = fs.createReadStream("recording.opus", { highWaterMark: CHUNK_SIZE });
  stream.on("data", (chunk) => ws.send(chunk));
  stream.on("end", () => ws.send(""));
});

ws.on("message", (data) => {
  const msg = JSON.parse(data.toString());
  if (msg.type === "utterance") {
    console.log(`[Speaker ${msg.utterance.speaker}] ${msg.utterance.text}`);
  } else if (msg.type === "done") {
    console.log(`Done. Duration: ${msg.duration_ms}ms`);
    ws.close();
  } else if (msg.type === "error") {
    console.error("Error:", msg.error);
    ws.close();
  }
});

ws.on("close", (code, reason) => {
  if (code !== 1000) console.error(`Closed unexpectedly: ${code} ${reason}`);
});

ws.on("error", (err) => console.error("WebSocket error:", err.message));

Error handling

async function safeRequest(url, options) {
  const response = await fetch(url, options);
  if (!response.ok) {
    let detail = response.statusText;
    try {
      const body = await response.json();
      detail = body.detail ?? detail;
    } catch {}

    const err = new Error(`API error ${response.status}: ${detail}`);
    err.status = response.status;
    throw err;
  }
  return response.json();
}

try {
  const result = await safeRequest(url, options);
} catch (err) {
  if (err.status === 429) {
    console.warn("Rate limited — wait and retry");
  } else if (err.status >= 500) {
    console.error("Server error — retry after delay");
  } else {
    throw err;
  }
}

WebSocket testing

WebSocket APIs cannot be tested with cURL. Use websocat for command-line testing.

# Install
cargo install websocat

# Basic connection test (Multilingual Transcription streaming)
websocat "wss://platform.modulate.ai/api/velma-2-stt-streaming?api_key=YOUR_API_KEY&speaker_diarization=true"

Get started

By capability

Guides

Code examples by language

cURL

Batch file upload with optional features

Deepfake Detection batch — minimal example

Python

Setup

Authentication — synchronous (requests)

Authentication — async (aiohttp)

Batch file upload (aiohttp)

Concurrent batch processing

WebSocket streaming (aiohttp)

Error handling

JavaScript (Node.js)

Setup

Batch file upload (native fetch, Node 18+)

Batch file upload (form-data package, Node < 18 or streams)

WebSocket streaming (ws package)

Error handling

WebSocket testing

​cURL

​Batch file upload with optional features

​Deepfake Detection batch — minimal example

​Python

​Setup

​Authentication — synchronous (requests)

​Authentication — async (aiohttp)

​Batch file upload (aiohttp)

​Concurrent batch processing

​WebSocket streaming (aiohttp)

​Error handling

​JavaScript (Node.js)

​Setup

​Batch file upload (native fetch, Node 18+)

​Batch file upload (form-data package, Node < 18 or streams)

​WebSocket streaming (ws package)

​Error handling

​WebSocket testing

​Related

cURL

Batch file upload with optional features

Deepfake Detection batch — minimal example

Python

Setup

Authentication — synchronous (requests)

Authentication — async (aiohttp)

Batch file upload (aiohttp)

Concurrent batch processing

WebSocket streaming (aiohttp)

Error handling

JavaScript (Node.js)

Setup

Batch file upload (native fetch, Node 18+)

Batch file upload (form-data package, Node < 18 or streams)

WebSocket streaming (ws package)

Error handling

WebSocket testing

Related