Tutorial · Tutorial · Python

Batch Vocal Separation with the API

One song is a demo. A thousand songs is a pipeline. This tutorial walks through a production-grade batch vocal separation flow: submit all jobs, listen for webhooks, collect the stems.

PythonBatchVocal separation

Step 1 — Prepare the batch manifest

Each input is an object-storage URL (S3/GCS/OSS). Build a manifest of every track you want separated.

Python
import json

manifest = [
    "s3://bucket/music/album1/track01.mp3",
    "s3://bucket/music/album1/track02.mp3",
    # ... 1,000 tracks
]

with open("manifest.json", "w") as f:
    json.dump(manifest, f)

Step 2 — Submit every job

Python
import requests

API = "https://api.mlslabs.io/v1"
HEADERS = {"X-API-Key": "YOUR_API_KEY"}

job_ids = []
for url in manifest:
    r = requests.post(
        API + "/vocal-separator/jobs",
        headers=HEADERS,
        json={
            "input": url,
            "output": "s3://bucket/music/stems/",
            "mode": "music",
            "tracks": ["vocals", "instrumental"],
            "webhook": "https://your-app.example.com/hooks/mlslabs",
        },
    ).json()
    job_ids.append(r["job_id"])

print(f"submitted {len(job_ids)} jobs")

Step 3 — Listen for webhooks

Python
# Flask/FastAPI endpoint — run this on your server
from flask import Flask, request

app = Flask(__name__)

@app.post("/hooks/mlslabs")
def on_job_done():
    payload = request.json
    if payload["status"] == "done":
        print("OK:", payload["job_id"], payload["output"]["vocals"])
    elif payload["status"] == "failed":
        print("FAILED (auto-refunded):", payload["job_id"])
    return {"ok": True}

if __name__ == "__main__":
    app.run(port=8080)

Step 4 — Track progress & retry

Python
import requests

def status(job_id):
    return requests.get(
        API + "/vocal-separator/jobs/" + job_id,
        headers=HEADERS,
    ).json()["status"]

# poll the manifest's remaining jobs every 30s
pending = job_ids[:]
while pending:
    pending = [j for j in pending if status(j) in ("queued", "running")]
    time.sleep(30)
print("all done")

Notes for production

  • Set a per-job webhook and store job_ids for audit
  • Failed jobs auto-refund — log and retry them
  • Rate-limit submissions to match the free-tier quota
  • Use output prefixes per album for clean S3 layout

FAQ

Common questions

How fast is 1,000 songs?

Parallel workers scale with the queue — the wall-clock time for a thousand 3-minute songs is usually minutes to a few hours, not weeks.

What if a job fails halfway?

Credits are auto-refunded and the webhook reports the failure — log the job_id and resubmit.

Do I need S3?

Any object storage the API supports works (S3/GCS/OSS). You can also pass direct media URLs for smaller batches.