1
from twelvelabs import TwelveLabs
2
from twelvelabs.indexes import IndexesCreateRequestModelsItem
3
from twelvelabs.tasks import TasksRetrieveResponse
4
5
client = TwelveLabs(api_key="<YOUR_API_KEY>")
6
7
index = client.indexes.create(
8
    index_name="<YOUR_INDEX_NAME>",
9
    models=[
10
        IndexesCreateRequestModelsItem(
11
            model_name="marengo2.7",
12
            model_options=["visual", "audio"]
13
        )
14
    ]
15
)
16
if index.id is None:
17
    raise RuntimeError("Failed to create an index.")
18
print(f"Created index: id={index.id}")
19
20
task = client.tasks.create(
21
    index_id=index.id, video_url="<YOUR_VIDEO_URL>")
22
print(f"Created task: id={task.id}")
23
24
def on_task_update(task: TasksRetrieveResponse):
25
    print(f"  Status={task.status}")
26
task = client.tasks.wait_for_done(task_id=task.id, callback=on_task_update)
27
if task.status != "ready":
28
    raise RuntimeError(f"Indexing failed with status {task.status}")
29
print(
30
    f"Upload complete. The unique identifier of your video is {task.video_id}.")
31
32
search_pager = client.search.query(
33
    index_id=index.id, query_text="<YOUR_QUERY>", search_options=["visual", "audio"],)
34
print("Search results:")
35
for clip in search_pager:
36
    print(
37
        f" video_id {clip.video_id} score={clip.score} start={clip.start} end={clip.end} confidence={clip.confidence}"
38
    )