> For the documentation index, fetch https://sync.so/docs/llms.txt. Append .md to a page URL for Markdown. Documentation-search MCP: https://sync.so/docs/_mcp/server.

# Personalized Video Messaging

> Create personalized video messages at scale using AI lip sync. Automate sales outreach and marketing with the sync. labs API and ElevenLabs.

## Personalized Video Messaging

Personalized video messaging uses AI lip sync to create unique video content for individual recipients at scale. By combining text-to-speech with Sync Labs' lip sync API, a single recorded video becomes thousands of personalized messages — each with the speaker appearing to address the recipient by name. This approach leads to higher engagement, better conversion rates, and a more personal connection compared to generic video messages.

This page covers two approaches: the text-to-speech example project below, and [replacing one word of the recording with each recipient's name](#replace-a-word-with-each-recipients-name-early-access) using dialogue edits and the Batch API.

Follow these steps to run the personalized video messaging example:

#### Clone Github Repository

Clone the [sync-examples repository](https://github.com/synchronicity-labs/sync-examples).

```bash
git clone https://github.com/synchronicity-labs/sync-examples.git
cd sync-examples/personalized-video-messsging/python
```

#### Set Up Environment

Create a virtual environment and install the required dependencies:

```bash
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
```

#### Configure API Keys

You will need [Sync\_API\_KEY](https://sync.so/settings/api-keys) and [ELEVEN\_LABS\_API\_KEY](https://elevenlabs.io/docs/api-reference/authentication#api-keys) to run the example. Update constants.py file with your own API keys:

* `SYNC_API_KEY`: Your API key for Sync Labs
* `ELEVEN_LABS_API_KEY`: Your API key for ElevenLabs (used for voice generation).

Ensure the file is saved after adding your keys.

#### Prepare Input Data (Optional)

The repository includes a sample input file `example.csv` for quickstart.
You can modify it with your own inputs if desired. Each row typically represents one recipient.

The input file should have the following columns:

* `video`: URL of the video to be personalized.
* `text`: Text to be personalized in the video.
* `segment_start`: start time of the video segment to be personalized.
* `segment_end`: end time of the video segment to be personalized.
* `output_format`: output format of the video.
* `sync_mode`: mode to sync text to video segment if lengths don't match. [options](/developer-guides/sync-mode)
* `voice_id`: elevenlabs voice id to use for the video. If empty, audio from the video will be cloned.
* `lipsync_model`: sync.so lipsync model to use for the video. default: `lipsync-2`
* `tts_model`: elevenlabs text to speech model. default: `eleven_multilingual_v2`

#### Generate Videos

Execute the Python script:

```bash
python main.py
```

This command will start the process of generating personalized videos based on the API keys you configured and the data in `example.csv`.

#### Check Results

On completion, the script will save the results to `outputs.csv` within the same directory. This file will contain output urls corresponding to each input.

That's it! You've now created personalized video messages for your customers.

## Replace a word with each recipient's name (early access)

Instead of generating new audio for the whole message, you can keep the recording and change one word, such as a name, for each recipient. Sync transcribes the video and re-speaks only that word in the speaker's cloned voice, then the [Batch API](/api-reference/guides/batch-processing) lip syncs every variant in one batch.

> **Note**
>
> This uses [dialogue edit rows](/api-reference/guides/batch-processing#dialogue-edit-rows-early-access) in the Batch API, which are in early access and enabled per organization. [Contact us](mailto:support@sync.so) to turn them on. The Batch API requires the Scale plan or higher.

You need:

* A single-speaker video uploaded to Sync storage. Follow [Asset Uploads](/developer-guides/asset-uploads) and keep the hosted URL.
* The word to replace, as spoken in the video. For example, record the message with a placeholder name such as "Alex" and replace that word.
* A CSV with a `name` column and 20 to 500 rows, one per recipient. An optional `pronunciation` column holds a respelling for names the voice says wrong. The first rows might look like this:

**`names.csv`**

```csv names.csv
name,pronunciation
Maya,
Siobhan,shivawn
Liam,
```

### How it works

1. Transcribe the video with `POST /v2/transcriptions` and find the `id` of the word to replace.
2. Create one preview per CSV row with `POST /v2/dialogue-edits`, changing that word to the row's name. The first preview clones the voice and returns `voiceId`; pass it as `voiceId` on the other previews to skip cloning. Preview requests are limited to 30 per minute, so space them about 2 seconds apart.
3. Wait for every preview to complete. Listen to the `previewAudioUrl` of a few previews before a large batch. If the voice says a name wrong, add a `pronunciation` respelling, for example `{"kind": "change", "wordId": "w35", "replacement": "Mia", "pronunciation": "Meeah"}`. The transcript and results keep the `replacement` text.
4. Submit one batch row per completed preview, first with `dry_run: true` and then for real. Every row sends the same source video and `dialogueEdit: {"id": "..."}`. Each row is charged the same as rendering that preview with `POST /v2/generate`.
5. Poll `GET /v2/batch/{id}` until it is `COMPLETED`, or pass `webhook_url` to be notified. Download the results file from `outputUrl` and match each line to its CSV row by `request_id`. A completed row's video is at `response.output_url`.

### Run the example script

The script below runs all five steps with the Python 3 standard library (3.8 or later, nothing to install). Save it as `personalize_batch.py`, then run:

```bash
export SYNC_API_KEY=your-api-key
python3 personalize_batch.py \
  --video-url "https://your-sync-hosted-video-url.mp4" \
  --csv names.csv \
  --word Alex > results.tsv
```

Progress goes to stderr, including each preview's audio URL so you can listen to a few. A preview that completes only partially (`COMPLETED_PARTIAL`) is still used, with a note to listen to it first; rows whose preview fails or times out are reported and left out of the batch. The script never resubmits a batch: if the submit returns a 5xx error or times out, the batch may exist, so it tells you to list your batches with `GET /v2/batch` and resume with `--batch-id`. Results go to stdout as tab-separated lines with the CSV line, name, status, and output URL (or error). Optional flags:

* `--dry-run` creates the previews and validates the batch, then stops without rendering any video.
* `--batch-id <id>` resumes an interrupted run: it polls that batch and prints its results. Pass the same CSV.
* `--model` sets the lip sync model (default `sync-3`).
* `--occurrence N` replaces the Nth match of `--word` when it is spoken more than once.
* `--webhook-url` sets the batch completion webhook.
* `--timeout-min N` limits how long each step (transcription, previews, batch) waits, 60 minutes by default. If the batch is still running when it expires, the script prints the `--batch-id` command to resume.

Set `SYNC_API_BASE` to use another API host.

**`personalize_batch.py`**

```python personalize_batch.py
#!/usr/bin/env python3
"""Personalize one video for every name in a CSV with dialogue edits and one batch.

Usage:
  export SYNC_API_KEY=your-api-key
  python3 personalize_batch.py --video-url URL --csv names.csv --word SaaS > results.tsv

The CSV needs a `name` column and 20 to 500 rows. An optional `pronunciation`
column holds a respelling the voice speaks instead of the name. Set
SYNC_API_BASE to use another API host. Python 3.8+, standard library only.
"""
import argparse
import csv
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request

BASE = os.environ.get("SYNC_API_BASE", "https://api.sync.so").rstrip("/")
API_KEY = os.environ.get("SYNC_API_KEY", "")
PREVIEW_GAP_SECONDS = 2.1  # previews allow 30 requests per minute
MIN_ROWS, MAX_ROWS = 20, 500
STARTED = time.time()


class Timeout(Exception):
    pass


class ApiError(Exception):
    def __init__(self, status, body):
        self.status = status
        self.body = body
        super().__init__(
            f"HTTP {status} {body.get('errorCode', '')}: {body.get('message', body)}"
        )


def log(message):
    print(f"[{time.time() - STARTED:4.0f}s] {message}", file=sys.stderr, flush=True)


def request(method, path, body=None, retry_statuses=(429,), retry_network=False):
    """Sends one API call and returns its JSON, waiting out the given statuses."""
    for attempt in range(6):
        req = urllib.request.Request(
            BASE + path,
            method=method,
            data=None if body is None else json.dumps(body).encode(),
            headers={
                "x-api-key": API_KEY,
                "Content-Type": "application/json",
                "User-Agent": "sync-personalize-batch-example/1.0",
            },
        )
        try:
            with urllib.request.urlopen(req, timeout=120) as res:
                return json.load(res)
        except urllib.error.HTTPError as err:
            raw = err.read().decode(errors="replace")
            try:
                detail = json.loads(raw)
            except ValueError:
                detail = {"message": raw[:300]}
            if err.code not in retry_statuses or attempt == 5:
                raise ApiError(err.code, detail)
            retry_after = err.headers.get("Retry-After", "")
            wait = float(retry_after) if retry_after.isdigit() else 5 * 2**attempt
        except OSError as err:
            # A POST that timed out may still have been accepted, so only reads retry.
            if not retry_network or attempt == 5:
                raise
            wait = 5 * 2**attempt
            log(f"  network error ({err}), retrying in {wait:.0f}s")
        time.sleep(wait)


def get(path):
    return request("GET", path, retry_statuses=(429, 500, 502, 503, 504), retry_network=True)


def post(path, body):
    return request("POST", path, body)


def wait_for(path, done, deadline, interval=5):
    while True:
        job = get(path)
        if job["status"] in done:
            return job
        if time.monotonic() > deadline:
            raise Timeout(f"{job.get('id')} is still {job['status']}")
        time.sleep(interval)


def read_recipients(path):
    with open(path, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        if "name" not in (reader.fieldnames or []):
            sys.exit(f"{path} needs a 'name' column")
        recipients = []
        for row in reader:
            name = (row.get("name") or "").strip()
            if not name:
                sys.exit(f"{path} line {reader.line_num}: 'name' is empty")
            pronunciation = (row.get("pronunciation") or "").strip()
            recipients.append(
                {"line": reader.line_num, "name": name, "pronunciation": pronunciation}
            )
    if not MIN_ROWS <= len(recipients) <= MAX_ROWS:
        sys.exit(f"A batch needs {MIN_ROWS} to {MAX_ROWS} rows; {path} has {len(recipients)}")
    return recipients


def normalize(text):
    return re.sub(r"\W", "", text).lower()


def transcribe(video_url, deadline):
    log("Transcribing the video")
    job = post("/v2/transcriptions", {"sourceVideoUrl": video_url})
    try:
        job = wait_for(f"/v2/transcriptions/{job['id']}", {"COMPLETED", "FAILED"}, deadline)
    except Timeout as err:
        sys.exit(f"Transcription {err}; rerun later, the same request reuses this job")
    if job["status"] == "FAILED":
        sys.exit(f"Transcription failed: {job.get('error')}")
    transcript = job["transcript"]
    speakers = transcript.get("speakerCount") or job.get("speakerCount") or 1
    if speakers > 1:
        sys.exit("Dialogue edits support single-speaker videos only")
    return transcript


def find_word(transcript, word, occurrence):
    words = [w for segment in transcript["segments"] for w in segment["words"]]
    target = normalize(word)
    if not target or " " in word.strip():
        sys.exit("--word must be a single word from the transcript")
    matches = [w for w in words if normalize(w["text"]) == target]
    if len(matches) < occurrence:
        text = " ".join(w["text"] for w in words)
        sys.exit(f"Found {len(matches)} occurrence(s) of '{word}' in: {text}")
    match = matches[occurrence - 1]
    context = " ".join(f"[{w['text']}]" if w is match else w["text"] for w in words)
    log(f"Replacing word {match['id']}: {context}")
    return match["id"]


def create_preview(video_url, transcript, word_id, recipient, voice_id):
    edit = {"kind": "change", "wordId": word_id, "replacement": recipient["name"]}
    if recipient["pronunciation"]:
        edit["pronunciation"] = recipient["pronunciation"]
    body = {"sourceVideoUrl": video_url, "transcript": transcript, "edits": [edit]}
    if voice_id:
        body["voiceId"] = voice_id
    return post("/v2/dialogue-edits", body)["id"]


def finish_preview(recipient, deadline):
    try:
        job = wait_for(
            f"/v2/dialogue-edits/{recipient['preview_id']}",
            {"COMPLETED", "COMPLETED_PARTIAL", "FAILED"},
            deadline,
        )
    except Timeout as err:
        recipient["error"] = f"preview {err} (check GET /v2/dialogue-edits/{recipient['preview_id']})"
        return {}
    if job["status"] == "FAILED":
        error = job.get("error") or {}
        recipient["error"] = f"preview {job['id']} FAILED {error.get('code', '')}".strip()
        return job
    recipient["preview"] = job["id"]
    log(f"  CSV line {recipient['line']} ({recipient['name']}): {job.get('previewAudioUrl')}")
    if job["status"] == "COMPLETED_PARTIAL":
        log(f"  note: CSV line {recipient['line']} is COMPLETED_PARTIAL (not every edit applied); listen first")
    return job


def make_previews(recipients, video_url, transcript, word_id, deadline):
    """Creates one preview per name; the first clones the voice, the rest reuse it."""
    first, rest = recipients[0], recipients[1:]
    log(f"Preview 1/{len(recipients)}: cloning the voice")
    first["preview_id"] = create_preview(video_url, transcript, word_id, first, None)
    voice_id = finish_preview(first, deadline).get("voiceId")
    if "preview" not in first:
        sys.exit(f"The first preview did not complete ({first['error']}); fix it before continuing")
    for i, recipient in enumerate(rest, start=2):
        time.sleep(PREVIEW_GAP_SECONDS)
        try:
            recipient["preview_id"] = create_preview(
                video_url, transcript, word_id, recipient, voice_id
            )
        except ApiError as err:
            recipient["error"] = str(err)
        if i % 10 == 0 or i == len(recipients):
            log(f"Submitted {i}/{len(recipients)} previews")
    for recipient in rest:
        if "preview_id" in recipient:
            finish_preview(recipient, deadline)
    ready = sum("preview" in r for r in recipients)
    log(f"{ready}/{len(recipients)} previews ready")


def refusal(err, rows, by_request_id):
    """Describes a refused batch and maps the failing line back to its CSV row."""
    body = err.body
    # Dialogue edit refusals carry lineNumber; other record errors name the line in message.
    line = body.get("lineNumber")
    if not isinstance(line, int):
        found = re.search(r"line (\d+)", str(body.get("message", "")))
        line = int(found.group(1)) if found else None
    text = f"HTTP {err.status} {body.get('errorCode', '')} lineNumber={line}: {body.get('message', body)}"
    if line and 1 <= line <= len(rows):
        recipient = by_request_id[rows[line - 1]["request_id"]]
        text += f" (CSV line {recipient['line']}, {recipient['name']})"
    return text


def build_rows(recipients, video_url, model):
    return [
        {
            "request_id": f"line-{r['line']}",
            "endpoint": "/v2/generate",
            "payload": {
                "model": model,
                "input": [{"type": "video", "url": video_url}],
                "dialogueEdit": {"id": r["preview"]},
            },
        }
        for r in recipients
        if "preview" in r
    ]


def download_results(output_url):
    # The result URL redirects to a signed file, so send no API key with it.
    req = urllib.request.Request(output_url, headers={"User-Agent": "sync-personalize-batch-example/1.0"})
    with urllib.request.urlopen(req, timeout=120) as res:
        return [json.loads(line) for line in res.read().decode().splitlines() if line.strip()]


def main():
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--video-url", help="Sync-hosted URL of the source video")
    parser.add_argument("--csv", required=True, help="CSV with a 'name' column")
    parser.add_argument("--word", help="Word in the video to replace with each name")
    parser.add_argument("--occurrence", type=int, default=1, help="Which match of --word to replace")
    parser.add_argument("--model", default="sync-3")
    parser.add_argument("--webhook-url", help="Optional batch completion webhook")
    parser.add_argument("--dry-run", action="store_true", help="Validate the batch without rendering")
    parser.add_argument("--batch-id", help="Resume: poll this batch and print its results")
    parser.add_argument("--timeout-min", type=float, default=60, help="Minutes to wait for each step (default 60)")
    args = parser.parse_args()
    if not API_KEY:
        sys.exit("Set SYNC_API_KEY")
    if not args.batch_id and not (args.video_url and args.word):
        sys.exit("--video-url and --word are required unless you pass --batch-id")
    if args.occurrence < 1 or args.timeout_min <= 0:
        sys.exit("--occurrence must be 1 or more and --timeout-min above 0")
    timeout = args.timeout_min * 60

    recipients = read_recipients(args.csv)
    by_request_id = {f"line-{r['line']}": r for r in recipients}
    batch_id = args.batch_id
    if not batch_id:
        transcript = transcribe(args.video_url, time.monotonic() + timeout)
        word_id = find_word(transcript, args.word, args.occurrence)
        make_previews(recipients, args.video_url, transcript, word_id, time.monotonic() + timeout)
        for r in recipients:
            if "error" in r:
                log(f"  skipping CSV line {r['line']} ({r['name']}): {r['error']}")
        rows = build_rows(recipients, args.video_url, args.model)
        if len(rows) < MIN_ROWS:
            sys.exit(f"Only {len(rows)} previews completed; a batch needs at least {MIN_ROWS}")

        try:
            post("/v2/batch", {"rows": rows, "dry_run": True})
        except ApiError as err:
            sys.exit(f"Dry run failed: {refusal(err, rows, by_request_id)}")
        log(f"Dry run validated {len(rows)} rows")
        if args.dry_run:
            return
        body = {"rows": rows}
        if args.webhook_url:
            body["webhook_url"] = args.webhook_url
        try:
            # Never retried: if the first attempt was accepted, a retry is a second batch.
            batch_id = request("POST", "/v2/batch", body, retry_statuses=())["id"]
        except (ApiError, OSError) as err:
            if isinstance(err, ApiError) and err.status < 500:
                sys.exit(f"Batch refused: {refusal(err, rows, by_request_id)}")
            sys.exit(
                f"Batch submit outcome unknown ({err}). The batch may have been created: "
                "list your batches with GET /v2/batch and resume with --batch-id <id> "
                "instead of submitting again."
            )
        log(f"Batch {batch_id} submitted; resume with --batch-id {batch_id}")

    progress, deadline = None, time.monotonic() + timeout
    while True:
        batch = get(f"/v2/batch/{batch_id}")
        metrics = batch.get("metrics") or {}
        if (batch["status"], metrics) != progress:
            progress = (batch["status"], metrics)
            log(f"Batch {batch['status']}: {json.dumps(metrics)}")
        if batch["status"] in ("COMPLETED", "FAILED"):
            break
        if time.monotonic() > deadline:
            sys.exit(f"Batch {batch_id} is still {batch['status']}; resume later with --batch-id {batch_id}")
        time.sleep(20)
    if not batch.get("outputUrl"):
        sys.exit(f"Batch {batch['status']} without a results file")

    # Results go to stdout as tab-separated lines; progress goes to stderr.
    print("csv_line\tname\tstatus\toutput_url_or_error")
    for line in download_results(batch["outputUrl"]):
        recipient = by_request_id.get(line["request_id"], {})
        response = line.get("response") or {}
        error = line.get("error") or ""
        if isinstance(error, dict):
            error = f"{error.get('code', '')} {error.get('message', '')}".strip()
        output = response.get("output_url") or response.get("outputUrl") or error
        print(f"{recipient.get('line', '')}\t{recipient.get('name', '')}\t{line['status']}\t{output}")


if __name__ == "__main__":
    main()
```

In a test with a 10-second clip and 20 names, the previews (one voice clone, then 19 reuses of its `voiceId`) were ready in about a minute and the batch completed all 20 rows about 5.5 minutes later. Batches have a target turnaround of 24 hours, so allow for longer runs, and use `--batch-id` to pick up where you left off. The results file starts like this:

**`results.tsv`**

```text results.tsv
csv_line	name	status	output_url_or_error
2	Olivia	COMPLETED	<output URL>
3	James	COMPLETED	<output URL>
```

## Frequently Asked Questions

#### How many personalized videos can I generate at once?

The CSV files above are input to the example scripts, not the Batch API. The [Batch API](/api-reference/guides/batch-processing) accepts 20 to 500 generation requests on eligible plans, as a JSONL file or as a JSON `rows` array. Each record must contain a unique `request_id`, `endpoint: "/v2/generate"`, and a generation `payload`. Convert your recipient data into that format before submitting a batch; for volumes above 500, split it into batches that each meet these limits.

#### What if the voice mispronounces a name?

With dialogue edits, add a `pronunciation` respelling to that row's `change` edit, for example "shivawn" for "Siobhan". Write it like an ordinary word, without hyphens or all-caps syllables, since capital letters may be spelled out one by one. Short names can still come out differently between takes. The voice speaks the respelling while the transcript and results keep the name. Listen to a few previews before you submit a large batch.

#### What TTS providers work with Sync Labs?

The example project includes built-in ElevenLabs integration for text-to-speech. However, any TTS provider that outputs standard WAV or MP3 audio files works with Sync Labs' lip sync API — simply pass the generated audio URL as the audio\_url parameter.

#### How much does personalized video messaging cost?

Cost depends on the lip sync model and video duration. Use the /v2/generate/estimate-cost endpoint to get exact pricing before submitting a batch. lipsync-2 is the most cost-effective option for high-volume personalized messaging. With dialogue edits, each batch row is charged the same as rendering that preview with `POST /v2/generate`.