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