TwelveLabs Python SDK

The TwelveLabs Python SDK provides a set of intuitive classes and methods that streamline platform interaction, minimizing the need for boilerplate code.

Note: The examples in this guide show only the required parameters. For the complete guides, see the Search, Analyze videos, and Create embeddings pages.

Prerequisites

Ensure that the following prerequisites are met before using the SDK:

• Python 3.7 or newer must be installed on your machine.

• To use the platform, you need an API key:

  1. If you don't have an account, sign up for a free account.
  2. Go to the API Keys page.
  3. If you need to create a new key, select the Create API Key button. Enter a name and set the expiration period. The default is 12 months.
  4. Select the Copy icon next to your key to copy it to your clipboard.

• Your video files must meet the following requirements:

  • For this guide: Files up to 4 GB.
  • Model capabilities: See the complete requirements for Marengo and Pegasus for resolution, aspect ratio, and supported formats.

  For upload size limits and processing modes, see the Upload and processing methods page.

Install the SDK

Install the latest version of the twelvelabs package:

pip install twelvelabs

Initialize the SDK

1. Import the SDK into your application:

   import time
   from twelvelabs import TwelveLabs
   

2. Instantiate the SDK client with your API key:

   client = TwelveLabs(api_key="<YOUR_API_KEY>")
   

Use the SDK

Upload your video, then follow the section for your task: search, analyze, or create embeddings. See our documentation for the complete list of features the platform provides.

Upload a video

To upload a video, call the client.assets.create method with the following parameters:

• method: Upload method. Use "url" for publicly accessible URLs or "direct" for local files.
• url or file: The video URL or an opened file object in binary read mode.

asset = client.assets.create(
    method="url",
    url="<YOUR_VIDEO_URL>" # Use direct links to raw media files. Video hosting platforms and cloud storage sharing links are not supported
    # Or use method="direct" and file=open("<PATH_TO_VIDEO_FILE>", "rb") to upload a local file.
)
print(f"Created asset: id={asset.id}")

The client.assets.create method returns an object that includes, among other information, a field named id representing the unique identifier of your asset. Use this identifier in the sections that follow.

Check the status of the asset

You only need this step for files larger than 200 MB. The platform processes files up to 200 MB synchronously and sets the asset status to ready. For larger files, check the asset status until it is ready.

To check the status of the asset, call the client.assets.retrieve method with the unique identifier of your asset as a parameter:

print("Waiting for asset to be ready...")
while True:
    asset = client.assets.retrieve(asset.id)
    if asset.status == "ready":
        print("Asset is ready")
        break
    if asset.status == "failed":
        raise RuntimeError(f"Asset processing failed: id={asset.id}")
    time.sleep(5)

Search

Search finds matching moments within an index. Create an index, add your uploaded video to it, then run queries.

Create an index

Indexes store and organize your video data, allowing you to group related videos. When you create an index, configure which Marengo model processes your videos and which modalities it analyzes.

To create an index, call the client.indexes.create method with the following parameters:

• index_name: The name of the index.
• models: An array of models to enable. Each entry has two fields:
  • model_name: The model to enable. Use "marengo3.0" for search.
  • model_options: The modalities to analyze.

index = client.indexes.create(
    index_name="<YOUR_INDEX_NAME>",
    models=[
        {"model_name": "marengo3.0", "model_options": ["visual", "audio"]}
    ]
)
if not index.id:
    raise RuntimeError("Failed to create an index.")
print(f"Created index: id={index.id}")

The client.indexes.create method returns an object that includes, among other information, a field named id representing the unique identifier of your new index.

See the Indexes page for more details.

Add your video to the index

Add the video you uploaded to your index. Call the client.indexes.indexed_assets.create method with the following parameters:

• index_id: The unique identifier of your index.
• asset_id: The unique identifier of the asset to index.

indexed_asset = client.indexes.indexed_assets.create(
    index_id=index.id,
    asset_id=asset.id
)
print(f"Created indexed asset: id={indexed_asset.id}")

The client.indexes.indexed_assets.create method returns an object that includes, among other information, a field named id representing the unique identifier of your indexed asset.

Monitor the indexing process

The platform indexes videos asynchronously. To monitor the indexing process, call the client.indexes.indexed_assets.retrieve method with the following parameters:

• index_id: The unique identifier of your index.
• indexed_asset_id: The unique identifier of your indexed asset.

import time

print("Waiting for indexing to complete.")
while True:
    indexed_asset = client.indexes.indexed_assets.retrieve(
        index_id=index.id,
        indexed_asset_id=indexed_asset.id
    )
    print(f"  Status={indexed_asset.status}")
    if indexed_asset.status == "ready":
        print("Indexing complete!")
        break
    elif indexed_asset.status == "failed":
        raise RuntimeError("Indexing failed")
    time.sleep(5)

The client.indexes.indexed_assets.retrieve method returns an object that includes, among other information, a field named status representing the status of the indexing process. Poll this method until status is "ready" before you run queries.

Run a query

Use natural language, images, or both to find matching video segments.

Text queries

To search using a text query, call the client.search.query method with the following parameters:

• query_text: Natural language query. The maximum length of a query is 500 tokens.
• search_options: Modalities to search. Valid values: "visual", "audio", "transcription" (spoken words). See the Search options page for details.

search_results = client.search.query(
    index_id=index.id,
    query_text="<YOUR_QUERY>",
    search_options=["visual", "audio"]
)
for i, clip in enumerate(search_results):
    print(f"Result {i + 1}: video_id={clip.video_id} rank={clip.rank} start={clip.start}s end={clip.end}s")

The client.search.query method returns an iterable where each item contains, among other information, the following fields:

• video_id: The unique identifier of the matching video.
• rank: The relevance ranking (1 = most relevant).
• start, end: The start and end time of the matching clip, expressed in seconds.

Image queries

To search using an image query, call the client.search.query method with the following parameters:

• query_media_type: Must be "image".
• query_media_file, query_media_url, query_media_files, or query_media_urls: The image or images to use as a query (up to 10 total). Provide at least one of the following:
  • (Optional) query_media_file: An opened file object in binary read mode.
  • (Optional) query_media_url: The publicly accessible URL of your image file.
  • (Optional) query_media_files: A list of opened file objects in binary read mode.
  • (Optional) query_media_urls: A list of publicly accessible URLs.

search_results = client.search.query(
    index_id=index.id,
    query_media_type="image",
    query_media_url="<YOUR_IMAGE_URL>",
    # Or use query_media_file=open("<PATH_TO_IMAGE_FILE>", "rb") for a local file.
    # Or use query_media_urls=["<URL_1>", "<URL_2>"] for multiple URLs.
    # Or use query_media_files=[open("<FILE_1>", "rb"), open("<FILE_2>", "rb")] for multiple local files.
    search_options=["visual"]
)
for i, clip in enumerate(search_results):
    print(f"Result {i + 1}: video_id={clip.video_id} rank={clip.rank} start={clip.start}s end={clip.end}s")

The response is similar to that received when using text queries.

Composed queries

Combine up to 10 images with text to narrow results. For example, provide an image of a car and add "red color" to find only red instances of that vehicle.

To perform a composed query, call the client.search.query method with the following parameters:

• query_media_file, query_media_url, query_media_files, or query_media_urls: The image or images to use as a query (up to 10 total). Provide at least one of the following:
  • (Optional) query_media_file: An opened file object in binary read mode.
  • (Optional) query_media_url: The publicly accessible URL of your image file.
  • (Optional) query_media_files: A list of opened file objects in binary read mode.
  • (Optional) query_media_urls: A list of publicly accessible URLs.
• query_text: Text that refines the image query.

search_results = client.search.query(
    index_id=index.id,
    query_media_type="image",
    query_media_url="<YOUR_IMAGE_URL>",
    # Or use query_media_file=open("<PATH_TO_IMAGE_FILE>", "rb") for a local file.
    # Or use query_media_urls=["<URL_1>", "<URL_2>"] for multiple URLs.
    # Or use query_media_files=[open("<FILE_1>", "rb"), open("<FILE_2>", "rb")] for multiple local files.
    query_text="<YOUR_QUERY>",
    search_options=["visual"]
)
for i, clip in enumerate(search_results):
    print(f"Result {i + 1}: video_id={clip.video_id} rank={clip.rank} start={clip.start}s end={clip.end}s")

The response is similar to that received when using text queries.

Analyze videos

The platform uses a multimodal approach to analyze video content, processing visuals, sounds, spoken words, and on-screen text. Use a custom prompt to generate summaries, extract insights, answer questions, or produce structured output.

Migrating from Pegasus 1.2? Pegasus 1.5 analyzes an asset directly and does not use an index or a video_id. For upgrade instructions, see the Migrate from Pegasus 1.2 to Pegasus 1.5 guide.

Note the following about using these methods:

• Pegasus 1.5 analyzes the asset you uploaded. You do not need an index.
• Your prompts can be instructive or descriptive, or you can phrase them as questions.
• The maximum length of a prompt is 2,000 tokens.

Streaming responses

Streaming delivers text fragments in real-time. Use it for live transcription or when you need immediate output.

To analyze a video with streaming responses, call the client.analyze_stream method with the following parameters:

• model_name: The model to use. Use "pegasus1.5".
• video: A VideoContext_AssetId object that identifies the asset to analyze. Set asset_id to the unique identifier of your asset.
• prompt_v_2: An AnalyzePromptV2 object. Set input_text to your prompt. The maximum length is 2,000 tokens.

from twelvelabs.types import VideoContext_AssetId, AnalyzePromptV2

text_stream = client.analyze_stream(
    model_name="pegasus1.5",
    video=VideoContext_AssetId(asset_id=asset.id),
    prompt_v_2=AnalyzePromptV2(input_text="<YOUR_PROMPT>")
)
for text in text_stream:
    if text.event_type == "text_generation":
        print(text.text)

The client.analyze_stream method returns a stream of objects. The event_type field can be "stream_start", "text_generation", or "stream_end".

Non-streaming responses

Non-streaming returns the complete text in a single response. Use it for reports or summaries where you need the full result at once. Call the client.analyze method with the same parameters as client.analyze_stream.

from twelvelabs.types import VideoContext_AssetId, AnalyzePromptV2

result = client.analyze(
    model_name="pegasus1.5",
    video=VideoContext_AssetId(asset_id=asset.id),
    prompt_v_2=AnalyzePromptV2(input_text="<YOUR_PROMPT>")
)
print(result.data)

The client.analyze method returns an object where the data field contains the complete generated text string (up to 4,096 tokens).

Create embeddings

Embeddings are vector representations of your content. Create them from video, audio, images, documents, and text, then use them for similarity search, classification, clustering, recommendations, or Retrieval-Augmented Generation (RAG).

Embed a query

Embed the text and media you search with. The platform processes your request synchronously and returns the embedding in the response.

from twelvelabs.types import MultiInputRequest

response = client.embed.v_2.create(
    input_type="multi_input",
    model_name="marengo3.5",
    multi_input=MultiInputRequest(input_text="<YOUR_QUERY>")
)
print(f"Dimensions: {len(response.data[0].embedding)}")

The data field contains one embedding. To combine text with images, video, and audio in a single embedding, reference a media source from your text, or request an uncertainty vector, see the Embed a query guide.

Embed content at scale

Embed the files you search through. The platform processes each file asynchronously, one file per request, so poll each task until it is ready.

from twelvelabs.types import AsyncVideoInputRequest, MediaSource

task = client.embed.v_2.tasks.create(
    input_type="video",
    model_name="marengo3.5",
    video=AsyncVideoInputRequest(media_source=MediaSource(asset_id=asset.id))
)

print("Waiting for the embedding task to be ready...")
while True:
    task = client.embed.v_2.tasks.retrieve(task.id)
    if task.status == "ready":
        print(f"Embeddings: {len(task.data)}")
        break
    if task.status == "failed":
        raise RuntimeError(f"Embedding task failed: id={task.id}")
    time.sleep(5)

This example embeds a video. The method also accepts "audio", "image", and "document" for PDF files. For the request each type takes, and for segmentation and scope options, see the Embed content at scale guide.

Error Handling

The SDK includes a set of exceptions that are mapped to specific HTTP status codes, as shown in the table below:

The following example shows how you can handle specific HTTP errors in your application:

import os
from twelvelabs import TwelveLabs
from twelvelabs.errors import BadRequestError, NotFoundError

client = TwelveLabs(api_key=os.getenv("TWELVELABS_API_KEY"))
try:
    indexes = client.indexes.list()
    print(indexes)
except BadRequestError as e:
    print("Bad request.")
except NotFoundError as e:
    print("Not found.")
except Exception as e:
    print(f"An error occurred: {e}")

Contributing

This repository contains code that has been automatically generated from an OpenAPI specification. We are unable to merge direct code contributions to the SDK because the code generation tool overwrites manual changes with each new release.

To contribute, follow these steps:

1. Open an issue to discuss your proposed changes with our team.
2. If you would like to submit a proof of concept, create a pull request. We will review your pull request, but we cannot merge it.
3. We will transfer any approved changes to the repository where the code generation tool operates.

We welcome contributions to the README file. You can submit pull requests directly.