1
import requests
2
import time
3
4
API_KEY = "<YOUR_API_KEY>"
5
BASE_URL = "https://api.twelvelabs.io/v1.3"
6
HEADERS = {"x-api-key": API_KEY}
7
8
# Step 1: Upload a video and wait until the asset is ready
9
response = requests.post(
10
    f"{BASE_URL}/assets",
11
    headers=HEADERS,
12
    files=[("method", (None, "url")), ("url", (None, "<YOUR_VIDEO_URL>"))]
13
)
14
asset_id = response.json()["_id"]
15
print(f"Asset created: {asset_id}")
16
17
while True:
18
    status = requests.get(f"{BASE_URL}/assets/{asset_id}", headers=HEADERS).json()["status"]
19
    if status == "ready":
20
        break
21
    elif status == "failed":
22
        raise Exception("Asset processing failed")
23
    print(f"Status: {status}, waiting...")
24
    time.sleep(5)
25
print("Asset ready")
26
27
# Step 2: Create a knowledge store with an ingestion config (optional but recommended)
28
response = requests.post(
29
    f"{BASE_URL}/knowledge-stores",
30
    headers={**HEADERS, "Content-Type": "application/json"},
31
    json={
32
        "name": "My First Knowledge Store",
33
        "ingestion_config": {
34
            "enrichment_config": {
35
                "description": "Extract key topics, themes, people, locations, and notable events"
36
            }
37
        }
38
    }
39
)
40
store_id = response.json()["_id"]
41
print(f"Knowledge store created: {store_id}")
42
43
# Step 3: Add the video to the knowledge store and wait until indexing completes
44
response = requests.post(
45
    f"{BASE_URL}/knowledge-stores/{store_id}/items",
46
    headers={**HEADERS, "Content-Type": "application/json"},
47
    json={"asset_id": asset_id}
48
)
49
item_id = response.json()["_id"]
50
print(f"Item added: {item_id}")
51
52
while True:
53
    status = requests.get(
54
        f"{BASE_URL}/knowledge-stores/{store_id}/items/{item_id}",
55
        headers=HEADERS
56
    ).json()["status"]
57
    if status == "ready":
58
        break
59
    elif status == "failed":
60
        raise Exception("Indexing failed")
61
    print(f"Status: {status}, waiting...")
62
    time.sleep(10)
63
print("Indexing complete")
64
65
# Step 4: Generate a response
66
response = requests.post(
67
    f"{BASE_URL}/responses",
68
    headers={**HEADERS, "Content-Type": "application/json"},
69
    json={
70
        "model": "jockey1.0",
71
        "input": [
72
            {"type": "message", "role": "user", "content": "Give me an overview of this video collection - main themes, key subjects, and any patterns across the videos."}
73
        ],
74
        "knowledge_store_id": store_id
75
    }
76
)
77
result = response.json()
78
for output in result["output"]:
79
    if output["type"] == "message":
80
        for content in output["content"]:
81
            print(content["text"])