diff --git a/docs/my-website/docs/providers/bedrock.md b/docs/my-website/docs/providers/bedrock.md
index 9e22f67527e..a9ac85a7571 100644
--- a/docs/my-website/docs/providers/bedrock.md
+++ b/docs/my-website/docs/providers/bedrock.md
@@ -1683,6 +1683,131 @@ curl --location 'http://0.0.0.0:4000/chat/completions' \
+## TwelveLabs Pegasus - Video Understanding
+
+TwelveLabs Pegasus 1.2 is a video understanding model that can analyze and describe video content. LiteLLM supports this model through Bedrock's `/invoke` endpoint.
+
+| Property | Details |
+|----------|---------|
+| Provider Route | `bedrock/us.twelvelabs.pegasus-1-2-v1:0`, `bedrock/eu.twelvelabs.pegasus-1-2-v1:0` |
+| Provider Documentation | [TwelveLabs Pegasus Docs ↗](https://docs.twelvelabs.io/docs/models/pegasus) |
+| Supported Parameters | `max_tokens`, `temperature`, `response_format` |
+| Media Input | S3 URI or base64-encoded video |
+
+### Supported Features
+
+- **Video Analysis**: Analyze video content from S3 or base64 input
+- **Structured Output**: Support for JSON schema response format
+- **S3 Integration**: Support for S3 video URLs with bucket owner specification
+
+### Usage with S3 Video
+
+
+
+
+```python title="TwelveLabs Pegasus SDK Usage" showLineNumbers
+from litellm import completion
+import os
+
+# Set AWS credentials
+os.environ["AWS_ACCESS_KEY_ID"] = "your-aws-access-key"
+os.environ["AWS_SECRET_ACCESS_KEY"] = "your-aws-secret-key"
+os.environ["AWS_REGION_NAME"] = "us-east-1"
+
+response = completion(
+ model="bedrock/us.twelvelabs.pegasus-1-2-v1:0",
+ messages=[{"role": "user", "content": "Describe what happens in this video."}],
+ mediaSource={
+ "s3Location": {
+ "uri": "s3://your-bucket/video.mp4",
+ "bucketOwner": "123456789012", # 12-digit AWS account ID
+ }
+ },
+ temperature=0.2
+)
+
+print(response.choices[0].message.content)
+```
+
+
+
+
+
+**1. Add to config**
+
+```yaml title="config.yaml" showLineNumbers
+model_list:
+ - model_name: pegasus-video
+ litellm_params:
+ model: bedrock/us.twelvelabs.pegasus-1-2-v1:0
+ aws_access_key_id: os.environ/AWS_ACCESS_KEY_ID
+ aws_secret_access_key: os.environ/AWS_SECRET_ACCESS_KEY
+ aws_region_name: os.environ/AWS_REGION_NAME
+```
+
+**2. Start proxy**
+
+```bash title="Start LiteLLM Proxy" showLineNumbers
+litellm --config /path/to/config.yaml
+
+# RUNNING at http://0.0.0.0:4000
+```
+
+**3. Test it!**
+
+```bash title="Test Pegasus via Proxy" showLineNumbers
+curl --location 'http://0.0.0.0:4000/chat/completions' \
+ --header 'Authorization: Bearer sk-1234' \
+ --header 'Content-Type: application/json' \
+ --data '{
+ "model": "pegasus-video",
+ "messages": [
+ {
+ "role": "user",
+ "content": "Describe what happens in this video."
+ }
+ ],
+ "mediaSource": {
+ "s3Location": {
+ "uri": "s3://your-bucket/video.mp4",
+ "bucketOwner": "123456789012"
+ }
+ },
+ "temperature": 0.2
+ }'
+```
+
+
+
+
+### Usage with Base64 Video
+
+You can also pass video content directly as base64:
+
+```python title="Base64 Video Input" showLineNumbers
+from litellm import completion
+import base64
+
+# Read video file and encode to base64
+with open("video.mp4", "rb") as video_file:
+ video_base64 = base64.b64encode(video_file.read()).decode("utf-8")
+
+response = completion(
+ model="bedrock/us.twelvelabs.pegasus-1-2-v1:0",
+ messages=[{"role": "user", "content": "What is happening in this video?"}],
+ mediaSource={
+ "base64String": video_base64
+ },
+ temperature=0.2,
+)
+
+print(response.choices[0].message.content)
+```
+
+### Important Notes
+
+- **Response Format**: The model supports structured output via `response_format` with JSON schema
+
## Provisioned throughput models
To use provisioned throughput Bedrock models pass
- `model=bedrock/`, example `model=bedrock/anthropic.claude-v2`. Set `model` to any of the [Supported AWS models](#supported-aws-bedrock-models)
@@ -1743,6 +1868,8 @@ Here's an example of using a bedrock model with LiteLLM. For a complete list, re
| Meta Llama 2 Chat 70b | `completion(model='bedrock/meta.llama2-70b-chat-v1', messages=messages)` | `os.environ['AWS_ACCESS_KEY_ID']`, `os.environ['AWS_SECRET_ACCESS_KEY']`, `os.environ['AWS_REGION_NAME']` |
| Mistral 7B Instruct | `completion(model='bedrock/mistral.mistral-7b-instruct-v0:2', messages=messages)` | `os.environ['AWS_ACCESS_KEY_ID']`, `os.environ['AWS_SECRET_ACCESS_KEY']`, `os.environ['AWS_REGION_NAME']` |
| Mixtral 8x7B Instruct | `completion(model='bedrock/mistral.mixtral-8x7b-instruct-v0:1', messages=messages)` | `os.environ['AWS_ACCESS_KEY_ID']`, `os.environ['AWS_SECRET_ACCESS_KEY']`, `os.environ['AWS_REGION_NAME']` |
+| TwelveLabs Pegasus 1.2 (US) | `completion(model='bedrock/us.twelvelabs.pegasus-1-2-v1:0', messages=messages, mediaSource={...})` | `os.environ['AWS_ACCESS_KEY_ID']`, `os.environ['AWS_SECRET_ACCESS_KEY']`, `os.environ['AWS_REGION_NAME']` |
+| TwelveLabs Pegasus 1.2 (EU) | `completion(model='bedrock/eu.twelvelabs.pegasus-1-2-v1:0', messages=messages, mediaSource={...})` | `os.environ['AWS_ACCESS_KEY_ID']`, `os.environ['AWS_SECRET_ACCESS_KEY']`, `os.environ['AWS_REGION_NAME']` |
## Bedrock Embedding
diff --git a/litellm/__init__.py b/litellm/__init__.py
index 778d972c6eb..475efaf5c91 100644
--- a/litellm/__init__.py
+++ b/litellm/__init__.py
@@ -1234,6 +1234,9 @@ from .llms.bedrock.chat.invoke_transformations.amazon_mistral_transformation imp
from .llms.bedrock.chat.invoke_transformations.amazon_titan_transformation import (
AmazonTitanConfig,
)
+from .llms.bedrock.chat.invoke_transformations.amazon_twelvelabs_pegasus_transformation import (
+ AmazonTwelveLabsPegasusConfig,
+)
from .llms.bedrock.chat.invoke_transformations.base_invoke_transformation import (
AmazonInvokeConfig,
)
diff --git a/litellm/constants.py b/litellm/constants.py
index b5826b43bfe..a4de43486b6 100644
--- a/litellm/constants.py
+++ b/litellm/constants.py
@@ -856,6 +856,7 @@ BEDROCK_INVOKE_PROVIDERS_LITERAL = Literal[
"nova",
"deepseek_r1",
"qwen3",
+ "twelvelabs",
]
BEDROCK_EMBEDDING_PROVIDERS_LITERAL = Literal[
diff --git a/litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py b/litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py
new file mode 100644
index 00000000000..62e98f7472f
--- /dev/null
+++ b/litellm/llms/bedrock/chat/invoke_transformations/amazon_twelvelabs_pegasus_transformation.py
@@ -0,0 +1,280 @@
+"""
+Transforms OpenAI-style requests into TwelveLabs Pegasus 1.2 requests for Bedrock.
+
+Reference:
+https://docs.twelvelabs.io/docs/models/pegasus
+"""
+
+import json
+import time
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+import httpx
+
+import litellm
+from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.core_helpers import map_finish_reason
+from litellm.llms.base_llm.base_utils import type_to_response_format_param
+from litellm.llms.base_llm.chat.transformation import BaseConfig
+from litellm.llms.bedrock.chat.invoke_transformations.base_invoke_transformation import (
+ AmazonInvokeConfig,
+)
+from litellm.llms.bedrock.common_utils import BedrockError
+from litellm.types.llms.openai import AllMessageValues
+from litellm.types.utils import ModelResponse, Usage
+from litellm.utils import get_base64_str
+
+if TYPE_CHECKING:
+ from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj
+
+ LiteLLMLoggingObj = _LiteLLMLoggingObj
+else:
+ LiteLLMLoggingObj = Any
+
+
+class AmazonTwelveLabsPegasusConfig(AmazonInvokeConfig, BaseConfig):
+ """
+ Handles transforming OpenAI-style requests into Bedrock InvokeModel requests for
+ `twelvelabs.pegasus-1-2-v1:0`.
+
+ Pegasus 1.2 requires an `inputPrompt` and a `mediaSource` that either references
+ an S3 object or a base64-encoded clip. Optional OpenAI params (temperature,
+ response_format, max_tokens) are translated to the TwelveLabs schema.
+ """
+
+ def get_supported_openai_params(self, model: str) -> List[str]:
+ return [
+ "max_tokens",
+ "max_completion_tokens",
+ "temperature",
+ "response_format",
+ ]
+
+ def map_openai_params(
+ self,
+ non_default_params: dict,
+ optional_params: dict,
+ model: str,
+ drop_params: bool,
+ ) -> dict:
+ for param, value in non_default_params.items():
+ if param in {"max_tokens", "max_completion_tokens"}:
+ optional_params["maxOutputTokens"] = value
+ if param == "temperature":
+ optional_params["temperature"] = value
+ if param == "response_format":
+ optional_params["responseFormat"] = self._normalize_response_format(
+ value
+ )
+ return optional_params
+
+ def _normalize_response_format(self, value: Any) -> Any:
+ """Normalize response_format to TwelveLabs format.
+
+ TwelveLabs expects:
+ {
+ "jsonSchema": {...}
+ }
+
+ But OpenAI format is:
+ {
+ "type": "json_schema",
+ "json_schema": {
+ "name": "...",
+ "schema": {...}
+ }
+ }
+ """
+ if isinstance(value, dict):
+ # If it has json_schema field, extract and transform it
+ if "json_schema" in value:
+ json_schema = value["json_schema"]
+ # Extract the schema if nested
+ if isinstance(json_schema, dict) and "schema" in json_schema:
+ return {"jsonSchema": json_schema["schema"]}
+ # Otherwise use json_schema directly
+ return {"jsonSchema": json_schema}
+ # If it already has jsonSchema, return as is
+ if "jsonSchema" in value:
+ return value
+ # Otherwise return the dict as is
+ return value
+ return type_to_response_format_param(response_format=value) or value
+
+ def transform_request(
+ self,
+ model: str,
+ messages: List[AllMessageValues],
+ optional_params: dict,
+ litellm_params: dict,
+ headers: dict,
+ ) -> dict:
+ input_prompt = self._convert_messages_to_prompt(messages=messages)
+ request_data: Dict[str, Any] = {"inputPrompt": input_prompt}
+
+ media_source = self._build_media_source(optional_params)
+ if media_source is not None:
+ request_data["mediaSource"] = media_source
+
+ # Handle temperature and maxOutputTokens
+ for key in ("temperature", "maxOutputTokens"):
+ if key in optional_params:
+ request_data[key] = optional_params.get(key)
+
+ # Handle responseFormat - transform to TwelveLabs format
+ if "responseFormat" in optional_params:
+ response_format = optional_params["responseFormat"]
+ transformed_format = self._normalize_response_format(response_format)
+ if transformed_format:
+ request_data["responseFormat"] = transformed_format
+
+ return request_data
+
+ def _build_media_source(self, optional_params: dict) -> Optional[dict]:
+ direct_source = optional_params.get("mediaSource") or optional_params.get(
+ "media_source"
+ )
+ if isinstance(direct_source, dict):
+ return direct_source
+
+ base64_input = optional_params.get("video_base64") or optional_params.get(
+ "base64_string"
+ )
+ if base64_input:
+ return {"base64String": get_base64_str(base64_input)}
+
+ s3_uri = (
+ optional_params.get("video_s3_uri")
+ or optional_params.get("s3_uri")
+ or optional_params.get("media_source_s3_uri")
+ )
+ if s3_uri:
+ s3_location = {"uri": s3_uri}
+ bucket_owner = (
+ optional_params.get("video_s3_bucket_owner")
+ or optional_params.get("s3_bucket_owner")
+ or optional_params.get("media_source_bucket_owner")
+ )
+ if bucket_owner:
+ s3_location["bucketOwner"] = bucket_owner
+ return {"s3Location": s3_location}
+ return None
+
+ def _convert_messages_to_prompt(self, messages: List[AllMessageValues]) -> str:
+ prompt_parts: List[str] = []
+ for message in messages:
+ role = message.get("role", "user")
+ content = message.get("content", "")
+ if isinstance(content, list):
+ text_fragments = []
+ for item in content:
+ if isinstance(item, dict):
+ item_type = item.get("type")
+ if item_type == "text":
+ text_fragments.append(item.get("text", ""))
+ elif item_type == "image_url":
+ text_fragments.append("")
+ elif item_type == "video_url":
+ text_fragments.append("