1
from twelvelabs import TwelveLabs
2
from twelvelabs.core.api_error import ApiError
3
4
client = TwelveLabs(api_key="<YOUR_API_KEY>")
5
6
try:
7
    store = client.knowledge_stores.retrieve(knowledge_store_id="<YOUR_KNOWLEDGE_STORE_ID>")
8
except ApiError as exc:
9
    print(f"Status: {exc.status_code}")
10
    print(f"Body: {exc.body}")

1
import time
2
from twelvelabs import TwelveLabs
3
from twelvelabs.core.api_error import ApiError
4
5
client = TwelveLabs(api_key="<YOUR_API_KEY>")
6
7
def with_retry(call, max_retries=3, base_delay=1):
8
    """Call an SDK method with exponential backoff on 429 and 5xx errors."""
9
    for attempt in range(max_retries):
10
        try:
11
            return call()
12
        except ApiError as exc:
13
            status = exc.status_code
14
            if status == 429 or (status is not None and status >= 500):
15
                delay = base_delay * (2 ** attempt)
16
                print(f"Received {status}, retrying in {delay}s...")
17
                time.sleep(delay)
18
                continue
19
            raise  # Client errors are not retryable
20
    raise RuntimeError("Exhausted retries")
21
22
response = with_retry(lambda: client.responses.create(
23
    knowledge_store_id="<YOUR_KNOWLEDGE_STORE_ID>",
24
    input=[{"type": "message", "role": "user", "content": "<YOUR_PROMPT>"}],
25
))

1
import time
2
3
def wait_for_ready(fetch, interval=5, timeout=600):
4
    """Poll a resource until it is ready or failed.
5
6
    `fetch` returns the current resource, for example:
7
    lambda: client.assets.retrieve(asset_id=asset_id)
8
    """
9
    elapsed = 0
10
    while elapsed < timeout:
11
        resource = fetch()
12
        if resource.status == "ready":
13
            return resource
14
        elif resource.status == "failed":
15
            raise Exception(f"Resource failed: {resource.id}")
16
        print(f"  Status: {resource.status} ({elapsed}s elapsed)")
17
        time.sleep(interval)
18
        elapsed += interval
19
    raise TimeoutError(f"Not ready after {timeout}s")

1
# Wait for an asset
2
asset = wait_for_ready(
3
    lambda: client.assets.retrieve(asset_id="<YOUR_ASSET_ID>"), interval=5
4
)
5
6
# Wait for a knowledge store item (longer interval - indexing takes more time)
7
item = wait_for_ready(
8
    lambda: client.knowledge_store_items.retrieve(
9
        knowledge_store_id="<YOUR_KNOWLEDGE_STORE_ID>", item_id="<YOUR_ITEM_ID>"
10
    ),
11
    interval=10,
12
    timeout=600,
13
)

1
import concurrent.futures
2
3
def wait_for_items(client, store_id, item_ids):
4
    """Wait for multiple knowledge store items in parallel."""
5
    def wait_one(item_id):
6
        return wait_for_ready(
7
            lambda: client.knowledge_store_items.retrieve(
8
                knowledge_store_id=store_id, item_id=item_id
9
            ),
10
            interval=10,
11
        )
12
13
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
14
        futures = {executor.submit(wait_one, iid): iid for iid in item_ids}
15
        return {futures[f]: f.result() for f in concurrent.futures.as_completed(futures)}