VOD Dubbing Automation Example

Build an automated multi-video dubbing system with the MK.IO API and webhooks. Submit the dubbing jobs, then use webhooks to start track insertion when each job completes.

What you will build:

• A job submission script that creates temporary assets and submits dubbing jobs
• A webhook listener that automatically inserts dubbed tracks when jobs complete

Before starting you will need encoded source videos in separate assets

You need three English-language videos that are already encoded for streaming using the mp4-v4 format, each in its own asset with a .mpd manifest present (for example, video-001-encoded, video-002-encoded, video-003-encoded).

If you do not have these yet, follow the previous guide to upload, create assets, and run an encoding job before using this demo.

Prerequisites

Step 1: Install Python Packages

pip install requests flask python-dotenv

Step 2: Install ngrok

Why ngrok? (local webhooks):

Webhooks must call a publicly reachable HTTPS URL - localhost is not accessible from MK.IO’s servers.

ngrok creates a temporary public URL that tunnels to your local Flask server so MK.IO can POST job events to your machine during development.

Installing ngrok: macOS (Homebrew):

brew install ngrok/ngrok/ngrok

Windows (Chocolatey):

choco install ngrok

Linux:

1. Download from ngrok.com/download,
2. unzip, and add the binary to your PATH

Step 3: Create Environment File

Create a file named .env in your project directory:

Add these lines (replace with your actual values):

MKIO_API_TOKEN=your_jwt_here
MKIO_PROJECT_NAME=your_project_name
STORAGE_ACCOUNT_NAME=your_storage_account_name
WEBHOOK_SECRET=<YOUR_WEBHOOK_SECRET>

Step 4: Create Transforms

You need to create 4 transforms in your MK.IO project. These are reusable templates that define how to process videos.

Important: Refer to the VOD Dubbing Demo Guide section “Create Transforms” for the exact API requests. You need to create:

1. dubbing-multi-lang - The main dubbing transform (English → Spanish, German, French)
2. insert-spanish - Add Spanish audio track
3. insert-german - Add German audio track
4. insert-french - Add French audio track

Creating the Scripts

File 1: submit_jobs.py

This script creates temporary assets for dubbed audio and submits dubbing jobs.

Create a file named submit_jobs.py:

# This script simply creates temporary output assets to hold the dubbed audio results and submits dubbing jobs for each encoded video file.

import os
import requests
from dotenv import load_dotenv

load_dotenv()

# ===== Get environment variables =====
API_TOKEN = os.getenv('MKIO_API_TOKEN')
PROJECT_NAME = os.getenv('MKIO_PROJECT_NAME')
STORAGE_ACCOUNT = os.getenv('STORAGE_ACCOUNT_NAME')

# ===== Define api endpoint and headers =====
BASE_URL = f"https://app.mk.io/api/v1/projects/{PROJECT_NAME}/media"
HEADERS = {
    'Authorization': f'Bearer {API_TOKEN}',
    'Content-Type': 'application/json'
}

# ===== Define encoded video names - CHANGE THESE to your actual encoded video asset names =====
VIDEOS = [
    'video-001-encoded',
    'video-002-encoded',
    'video-003-encoded'
]

# ===== This is the first step: To create an asset for each video to contain the generated dubbed audio .mp4 files. =====
def create_dubbed_audio_asset(source_asset):
    asset_name = f"{source_asset}-dubbed-audio"
    url = f"{BASE_URL}/assets/{asset_name}"
    body = {
        "properties": {
            "description": f"Dubbed audio for {source_asset}",
            "storageAccountName": STORAGE_ACCOUNT
        }
    }
    response = requests.put(url, headers=HEADERS, json=body)

# If we get error 409, the asset already exists - this is fine we can still proceed.
    if response.status_code == 409:
        print(f"Asset already exists: {asset_name}")
    else:
        response.raise_for_status()
        print(f"Created asset: {asset_name}")
    return asset_name

# ===== Get the first mp4 files from the encoded asset to run the dubbing job on. They all contain the necessary audio files - lowest bitrate is fine. =====
def get_first_mp4(asset_name):
    url = f"{BASE_URL}/assets/{asset_name}/storage"
    response = requests.get(url, headers=HEADERS)

# If the asset has no storage container it will return error 400
    if response.status_code == 400:
        return None
    response.raise_for_status()
    storage = response.json()
    files = storage.get('spec', {}).get('files', [])

# Find the first .mp4 file
    for f in files:
        name = f.get('name') or f.get('path')
        if name and name.endswith('.mp4'):
            return name
    return None

# ===== Submit a dubbing job to the dubbing transform created during setup. In this example it is called 'dubbing-multi-lang'. =====
def submit_dubbing_job(source_asset, dubbed_asset, input_file):
    job_name = f"dub-{source_asset}"
    url = f"{BASE_URL}/transforms/dubbing-multi-lang/jobs/{job_name}"
    body = {
        "properties": {
            "description": f"Dub {source_asset}",
            "priority": "Normal",
            "input": {
                "@odata.type": "#Microsoft.Media.JobInputAsset",
                "assetName": source_asset,
                "files": [input_file]
            },
            "outputs": [{
                "@odata.type": "#Microsoft.Media.JobOutputAsset",
                "assetName": dubbed_asset
            }]
        }
    }
    response = requests.put(url, headers=HEADERS, json=body)
    response.raise_for_status()
    print(f"Submitted dubbing job: {job_name}")
    return job_name

# ===== The main entry point =====
def main():
    print("=================================")
    print(" MK.IO VOD Dubbing - submit jobs")
    print("=================================")
    print("Videos to process:")
    for video in VIDEOS:
        print(f" - {video}")
    print(f"\nCreating temporary assets for dubbed audio...\n")
    jobs_submitted = []
    for video in VIDEOS:
        # Create the output asset for dubbed audio
        dubbed_asset = create_dubbed_audio_asset(video)
        # Get the input file from the source asset
        input_file = get_first_mp4(video)
        if not input_file:
            print(f"  ERROR: No mp4 file found in asset: {video}.")
            continue
        print(f"  Input file for {video}: {input_file}")
        # Submit the dubbing job
        job_name = submit_dubbing_job(video, dubbed_asset, input_file)
        jobs_submitted.append(job_name)
        print()
    print("="*10 + "\n")
    print(f"\n✓ Submitted {len(jobs_submitted)} dubbing jobs\n")
    print("\nDubbing will process in the background.")
    print("="*10 + "\n")

if __name__ == "__main__":
    main()

Edit lines 23-26 and change VIDEOS to your actual video asset names

File 2: webhook_listener.py

This script receives webhooks from MK.IO and automatically handles track insertion.

Create a file named webhook_listener.py:

import os
import requests
from flask import Flask, request, jsonify
from dotenv import load_dotenv

load_dotenv()

# ===== Get environment variables =====
API_TOKEN = os.getenv('MKIO_API_TOKEN')
PROJECT_NAME = os.getenv('MKIO_PROJECT_NAME')
WEBHOOK_SECRET = os.getenv('WEBHOOK_SECRET')

app = Flask(__name__)

def get_dubbed_files(asset_name):
    url = f"{BASE_URL}/assets/{asset_name}/storage"
    response = requests.get(url, headers=HEADERS)
    if response.status_code == 400:
        return {}
    response.raise_for_status()
    storage = response.json()
    files = storage.get('spec', {}).get('files', [])
    dubbed = {}
    for f in files:
        path = f.get('path') or f.get('name')
        if path:
            if '_es-ES.mp4' in path:
                dubbed['es-ES'] = path
            elif '_de-DE.mp4' in path:
                dubbed['de-DE'] = path
            elif '_fr-FR.mp4' in path:
                dubbed['fr-FR'] = path
    return dubbed

def submit_track_insertion_job(dubbed_asset, target_asset, dubbed_file, language_code, transform_name):
    lang_short = language_code.split('-')[0]  # e.g., 'es' from 'es-ES'
    job_name = f"insert-{target_asset}-{lang_short}"
    url = f"{BASE_URL}/transforms/{transform_name}/jobs/{job_name}"
    body ={
        "properties": {
            "description": f"Insert {language_code} track",
            "priority": "Normal",
            "input": {
                "@odata.type": "#Microsoft.Media.JobInputAsset",
                "assetName": dubbed_asset,
                "files": [dubbed_file]
            },
            "outputs": [{
                "@odata.type": "#Microsoft.Media.JobOutputAsset",
                "assetName": target_asset
            }]
        }
    }
    response = requests.put(url, headers=HEADERS, json=body)
    response.raise_for_status()
    print(f"{job_name} submitted for {language_code}")

def insert_all_tracks(source_asset, dubbed_asset):
    print (f"
 Inserting dubbed tracks for asset: {source_asset}")
    dubbed_files = get_dubbed_files(dubbed_asset)
    if not dubbed_files:
        print("No dubbed files found.")
        return
    languages = {
        'es-ES': 'insert-spanish',
        'fr-FR': 'insert-french',
        'de-DE': 'insert-german'
    }
    for lang_code,transform in languages.items():
        if lang_code in dubbed_files:
            dubbed_file = dubbed_files[lang_code]
            submit_track_insertion_job(dubbed_asset, source_asset, dubbed_file, lang_code, transform)

@app.route('/webhook', methods=['POST'])
def webhook():
    auth = request.headers.get('Authorization', '')
    expected_auth = f"Bearer {WEBHOOK_SECRET}"
    if auth != expected_auth:
        print("!! Webhook received with invalid authorization !!")
        return jsonify({"error": "Unauthorized"}), 401
    event = request.get_json()
    if not event:
        return jsonify({"error": "There is no body"}), 400
    event_type = event.get('type')
    data = event.get('data',{})
    resource = data.get('resource',{})
    job_name = resource.get('name')
    job_state= data.get('state')
    print(f"\n Webhook received: {event_type}")
    print(f"   Job: {job_name}")
    print(f"   State: {job_state}")
    if job_state == 'Finished' and job_name.startswith('dub-'):
        source_asset = job_name.replace('dub-', '')
        dubbed_asset = f"{source_asset}-dubbed-audio"
        insert_all_tracks(source_asset, dubbed_asset)
    elif job_state == 'finished' and job_name.startswith('insert-'):
        parts = job_name.split('-')
        lang = parts [-1]
        print (f" Track insertion for language {lang} completed.")
    elif job_state =='error':
        print(f" !! Job {job_name} failed !! ")
    return jsonify({"status": "received"}), 200

@app.route('/health', methods=['GET'])
def health():
    return jsonify({"status": "ok"}), 200

if __name__ == '__main__':
    print("\n" + "="*60)
    print("MK.IO VOD Dubbing - Webhook Listener")
    print("="*60)
    print("\nListening on http://localhost:5000")
    print("Webhook endpoint: http://localhost:5000/webhook\n")
    print("Make sure you:")
    print("  1. Created all transforms (using the guide)")
    print("  2. Created webhook rule in MK.IO dashboard")
    print("  3. Ran submit_jobs.py to submit dubbing jobs")
    print("\nWaiting for webhooks...\n")
    print("="*60 + "\n")
    app.run(host='0.0.0.0', port=5000, debug=False)

Running the Complete Pipeline

Step 1: Set Up ngrok Tunnel

Open a new terminal and run:

gnrok http 5000

Step 2: Create Webhook Rule

Using Postman, make a PUT request:

PUT https://app.mk.io/api/v1/projects/YOUR_PROJECT_NAME/webhook/rules/dubbing-demo

Replace:

• YOUR_PROJECT_NAME with your project name
• YOUR_API_TOKEN with your API token

You should get 200 OK.

Step 3: Start the Webhook Listener

Open a terminal and run:

python webhook_listener.py

Step 4: Submit Dubbing Jobs

Open a new terminal and run:

python submit_jobs.py

Monitor the workflow

The webhook listener terminal will start showing live updates:

============================================================

Webhook received: MediaKind.JobStarted

Job: dub-video-001-encoded

State: Scheduled