`client.task.create()` fails with error `Error uploading video: Failed connection`

[opringle]

Expected Behavior

• The video is downloaded, index created and video uploaded to the index

Actual Behavior

• The script fails with error Error uploading video: Failed connection

Steps to Reproduce

• copy the below code snippets into local directory
• update the API key
• pip install -r requirements.txt
• python twelvelabs_mwe.py

.env

TWELVELABS_API_KEY=<YOUR API KEY>

requirements.txt

python_dotenv==1.1.0
yt-dlp==2025.5.22
twelvelabs==0.4.10

twelvelabs_mwe.py

import os
from pathlib import Path
from dotenv import load_dotenv
from twelvelabs import APIStatusError, TwelveLabs
from twelvelabs.models.task import Task
import yt_dlp

load_dotenv()
api_key = os.getenv("TWELVELABS_API_KEY")
client = TwelveLabs(api_key=api_key)


# 1. Download the youtube video to local path
youtube_url = "https://www.youtube.com/watch?v=luP751orGiA&ab_channel=GoProBike"
track_id = "val_di_sole_2024"
rider_id = "loris_vergier"
video_dir = Path("downloaded_videos") / track_id / rider_id
video_dir.mkdir(parents=True, exist_ok=True)
video_path = video_dir / f'{track_id}_{rider_id}.mp4'
if video_path.exists():
    print(
        f"Video file already exists at {video_path}, skipping download.")
else:
    try:
        ydl_opts = {
            'format': 'bestvideo+bestaudio/best',
            'merge_output_format': 'mp4',
            'outtmpl': str(video_path),
        }
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            ydl.download([youtube_url])
        print(f"Downloaded video to {video_path}")
    except Exception as e:
        print(f"Error downloading video: {e}")
        exit(1)

# 2. create an index if it doesn't already exist
index_name = "test_index"
try:
    index = client.index.create(
        name=index_name,
        models=[{"name": "marengo2.7", "options": ["visual"]}]
    )
    index_id = index.id
    print(f"Created index with ID: {index_id}")
except APIStatusError as e:
    if e.status_code == 409:
        indexes = client.index.list()
        for idx in indexes:
            if idx.name == index_name:
                index_id = idx.id
                print(f"Using existing index with ID: {index_id}")
                break
        else:
            raise ValueError(f"Index {index_name} not found after 409 error")
    else:
        raise
except Exception as e:
    print(f"Unexpected error: {e}")
    exit(1)

# 3. upload the video to the index
if not os.path.exists(video_path):
    raise FileNotFoundError(f"Video file not found: {video_path}")

try:
    print(f"creating task...")
    task = client.task.create(
        index_id=index_id,
        file=video_path,
    )
    print(f"created task with id {task.id}")

    def callback_fn(task: Task):
        print(f"Task {task.id} status: {task.status}")
    task.wait_for_done(
        sleep_interval=2,
        callback=callback_fn
    )
    if task.status == "ready":
        print("Video uploaded successfully")
    else:
        print(f"Failed to upload video. Task status: {task.status}")
except Exception as e:
    print(f"Error uploading video: {e}")