This commit is contained in:
Jan Scherbaum 2026-10-05 08:24:27 +02:00 • committed by GitHub
commit 2795db90d6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 6187 additions and 33 deletions

View file

@ -749,7 +749,11 @@ def image_edit(
"n",
"quality",
"size",
"style",
# "style" is intentionally NOT blocklisted here: image_edit has no named
# `style` argument, so blocklisting it would drop the param for every
# provider. It only reaches handlers whose image-edit config advertises
# "style" in get_supported_openai_params (e.g. bedrock Nova Canvas,
# recraft); every other provider's transform ignores unknown keys.
"async_call",
]
non_default_params: Final = filter_out_litellm_params(kwargs, excluding=openai_params)
@ -962,9 +966,9 @@ def image_edit(
@client
async def aimage_edit(
image: FileTypes | list[FileTypes],
model: str,
prompt: str,
image: FileTypes | list[FileTypes] | None = None,
model: str | None = None,
prompt: str | None = None,
mask: str | None = None,
n: int | None = None,
quality: str | ImageGenerationRequestQuality | None = None,
@ -999,10 +1003,15 @@ async def aimage_edit(
# get custom llm provider so we can use this for mapping exceptions
if custom_llm_provider is None:
_, custom_llm_provider, _, _ = litellm.get_llm_provider(
model=model, api_base=local_vars.get("base_url", None)
model=model or DEFAULT_IMAGE_ENDPOINT_MODEL,
api_base=local_vars.get("base_url", None),
)
images: Final = image if isinstance(image, list) else [image]
images: Final = (
image
if isinstance(image, list)
else ([image] if image is not None else []) # mutable-ok: single-image wrap like sync image_edit
)
func: Final = partial(
image_edit,

View file

@ -132,6 +132,39 @@ def merge_bedrock_aws_request_params(
return request_params
# Headers safe to surface in logs/callbacks: non-sensitive request metadata.
# Everything else (Authorization, X-Amz-Security-Token, X-Amz-Date, ...) is
# signature material and must never leave the request path.
_BEDROCK_LOGGING_SAFE_HEADERS: Final[frozenset[str]] = frozenset(
{
"content-type",
"content-length",
"host",
"accept",
"x-amzn-requestid",
"x-amzn-errortype",
}
)
_BEDROCK_REDACTED_HEADER_VALUE: Final = "[REDACTED]"
def redact_bedrock_headers_for_logging(headers: Mapping[str, str]) -> dict[str, str]:
"""Copy ``headers`` with every non-allowlisted value replaced by ``[REDACTED]``.
SigV4-signed Bedrock requests carry credentials in headers (Authorization,
X-Amz-Security-Token, X-Amz-Date); ``Logging.pre_call`` forwards
``additional_args["headers"]`` unmasked to logger_fn and custom
``log_pre_api_call`` callbacks, so handlers must pass this copy instead of
the prepared request headers. Key names are preserved (case-sensitively)
so log consumers keep seeing the full header shape; the allowlist match is
case-insensitive per HTTP header semantics. The sent request is untouched.
"""
return { # mutable-ok: redacted logging copy handed to the caller
key: (value if key.lower() in _BEDROCK_LOGGING_SAFE_HEADERS else _BEDROCK_REDACTED_HEADER_VALUE)
for key, value in headers.items()
}
def s3_static_key_pair(params: Mapping[str, object]) -> tuple[str, str] | None:
"""The s3_access_key_id / s3_secret_access_key pair when both are set, otherwise None."""
s3_access_key_id: Final = params.get("s3_access_key_id")

View file

@ -4,6 +4,9 @@ Amazon Nova Canvas image edit on Bedrock (InvokeModel).
Maps OpenAI-style image edit (image + prompt, optional mask) to Nova Canvas task types:
- With mask: INPAINTING (inPaintingParams per AWS docs)
- Without mask: IMAGE_VARIATION (imageVariationParams)
- TEXT_IMAGE: conditioned editing where the input image conditions layout via
textToImageParams.conditionImage + controlMode (CANNY_EDGE | SEGMENTATION) +
controlStrength (issue #39552)
Refs:
- https://docs.aws.amazon.com/nova/latest/userguide/image-gen-access.html
@ -38,6 +41,44 @@ else:
LiteLLMLoggingObj = Any
NOVA_CANVAS_CONTROL_MODES: Final[tuple[str, ...]] = ("CANNY_EDGE", "SEGMENTATION")
def _invalid_input_error(message: str) -> BedrockError:
"""400-class error for invalid caller input.
A plain ValueError would surface as APIConnectionError (500-class, which OpenAI
SDKs auto-retry); BedrockError(status_code=400) maps to BadRequestError instead
(same mapping the Nova Reel video config relies on).
"""
return BedrockError(status_code=400, message=message)
def _resolve_edit_image_b64(
image: FileTypes | None,
condition_image_b64: str | None,
task_type: str | None,
) -> str:
"""Base64 image for the task body: the multipart ``image`` wins; ``conditionImage``
(already encoded) backs TEXT_IMAGE when no multipart file was sent."""
if condition_image_b64 is not None and task_type != "TEXT_IMAGE":
# Checked before the multipart-image early return: image + conditionImage
# with a non-TEXT_IMAGE taskType must fail loudly instead of silently
# discarding the caller's conditionImage.
raise _invalid_input_error(
"Amazon Nova Canvas conditionImage is only supported with "
f"taskType=TEXT_IMAGE (conditioned editing); got taskType={task_type!r}."
)
if image is not None:
return _file_types_to_b64(image)
if task_type != "TEXT_IMAGE" or condition_image_b64 is None:
raise _invalid_input_error(
"Nova Canvas image edit requires an image input. Pass the multipart "
"`image` file, or a `conditionImage` for taskType=TEXT_IMAGE."
)
return condition_image_b64
def _nova_canvas_task_body(
*,
image_b64: str,
@ -48,6 +89,9 @@ def _nova_canvas_task_body(
task_type: str | None,
mask_prompt: str | None,
out_painting_mode: str | None,
control_mode: str | None = None,
control_strength: float | str | None = None,
style: str | None = None,
) -> dict[str, object]:
"""Build InvokeModel body task section (without imageGenerationConfig)."""
if task_type == "BACKGROUND_REMOVAL":
@ -77,6 +121,50 @@ def _nova_canvas_task_body(
"taskType": "OUTPAINTING",
"outPaintingParams": out_params,
}
if task_type == "TEXT_IMAGE":
# Conditioned editing: the input image guides layout/composition of the
# generated image via textToImageParams.conditionImage. SEGMENTATION
# controlMode derives a segmentation mask from the condition image;
# CANNY_EDGE (the AWS default) follows its prominent contours.
if mask_b64 is not None or mask_prompt is not None:
# AWS TEXT_IMAGE has no mask field; fail fast instead of silently
# dropping the caller's mask or maskPrompt.
raise _invalid_input_error(
"Amazon Nova Canvas TEXT_IMAGE (conditioned editing) does not support a "
"mask. Use INPAINTING or OUTPAINTING for mask-based editing workflows."
)
if control_mode is not None and control_mode not in NOVA_CANVAS_CONTROL_MODES:
raise _invalid_input_error(
f"Unsupported Amazon Nova Canvas controlMode: {control_mode!r}. Use one of {NOVA_CANVAS_CONTROL_MODES}."
)
control_strength_value: float | None = None
if control_strength is not None:
# Multipart form data delivers controlStrength as a string; coerce
# before the range check (raw strings would TypeError on <=).
try:
control_strength_value = float(control_strength)
except (TypeError, ValueError):
raise _invalid_input_error("Amazon Nova Canvas controlStrength must be a number in [0.0, 1.0].")
if not 0.0 <= control_strength_value <= 1.0:
raise _invalid_input_error(
f"Amazon Nova Canvas controlStrength must be between 0.0 and 1.0; got {control_strength_value!r}."
)
t2i_params: Final[dict[str, object]] = { # mutable-ok: optional conditioned-editing keys are set below
"text": text,
"conditionImage": image_b64,
}
if negative_text is not None:
t2i_params["negativeText"] = negative_text
if control_mode is not None:
t2i_params["controlMode"] = control_mode
if control_strength_value is not None:
t2i_params["controlStrength"] = control_strength_value
if style is not None:
t2i_params["style"] = style
return { # mutable-ok: InvokeModel JSON body is a plain dict
"taskType": "TEXT_IMAGE",
"textToImageParams": t2i_params,
}
# Honour explicit IMAGE_VARIATION even when a mask is present (mask is ignored
# for this task type; callers use INPAINTING when they want mask semantics).
if task_type == "IMAGE_VARIATION":
@ -95,9 +183,10 @@ def _nova_canvas_task_body(
# Explicit taskType must be INPAINTING or omitted from here on; anything else is invalid.
if task_type is not None and str(task_type).strip() != "":
if task_type != "INPAINTING":
raise ValueError(
raise _invalid_input_error(
f"Unsupported Amazon Nova Canvas taskType: {task_type!r}. "
"Use BACKGROUND_REMOVAL, OUTPAINTING, IMAGE_VARIATION, INPAINTING, "
"TEXT_IMAGE (conditioned editing via conditionImage/controlMode), "
"or omit taskType for automatic routing (mask → INPAINTING, else IMAGE_VARIATION)."
)
if mask_b64 is not None or mask_prompt is not None or task_type == "INPAINTING":
@ -109,7 +198,7 @@ def _nova_canvas_task_body(
if negative_text is not None:
in_params["negativeText"] = negative_text
if "maskPrompt" not in in_params and "maskImage" not in in_params:
raise ValueError(
raise _invalid_input_error(
"Amazon Nova Canvas INPAINTING requires either maskPrompt or maskImage "
"(use OpenAI mask= for maskImage, or pass maskPrompt in optional params). "
"See https://docs.aws.amazon.com/nova/latest/userguide/image-gen-req-resp-structure.html"
@ -248,9 +337,13 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
"cfgScale",
"seed",
"quality",
"style",
"taskType",
"maskPrompt",
"outPaintingMode",
"controlMode",
"controlStrength",
"conditionImage",
"imageGenerationConfig",
]
@ -314,7 +407,15 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
headers: dict,
) -> tuple[dict, Any]:
op: Final = dict(image_edit_optional_request_params)
image_b64: Final = _file_types_to_b64(image)
# conditionImage: alternative source for the TEXT_IMAGE condition image
# for callers that cannot send a multipart `image` file (e.g. plain JSON
# bodies). When both are supplied the multipart `image` field wins.
condition_image_raw: Final = op.pop("conditionImage", None)
condition_image_b64: Final[str | None] = (
_file_types_to_b64(condition_image_raw) if condition_image_raw is not None else None
)
task_type: Final = op.pop("taskType", None)
image_b64: Final[str] = _resolve_edit_image_b64(image, condition_image_b64, task_type)
mask_raw: Final = op.pop("mask", None)
mask_b64: str | None = None
@ -353,12 +454,12 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
if seed is not None:
image_generation_config["seed"] = seed
task_type: Final = op.pop("taskType", None)
if (prompt is None or prompt == "") and task_type in (
"INPAINTING",
"OUTPAINTING",
"TEXT_IMAGE",
):
raise ValueError(
raise _invalid_input_error(
f"Amazon Nova Canvas {task_type} requires a text prompt. Pass a non-empty `prompt` in your request."
)
text: Final = prompt if prompt is not None and prompt != "" else " "
@ -366,6 +467,19 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
similarity_strength: Final = op.pop("similarityStrength", None)
mask_prompt: Final = op.pop("maskPrompt", None)
out_painting_mode: Final = op.pop("outPaintingMode", None)
control_mode: Final = op.pop("controlMode", None)
control_strength: Final = op.pop("controlStrength", None)
style: Final = op.pop("style", None)
if (
control_mode is not None or control_strength is not None or style is not None
) and task_type != "TEXT_IMAGE":
# Conditioning fields only exist on textToImageParams (TEXT_IMAGE);
# any other resolved task type would silently drop them.
raise _invalid_input_error(
"Amazon Nova Canvas controlMode/controlStrength/style are only supported "
f"with taskType=TEXT_IMAGE (conditioned editing); resolved taskType={task_type!r} "
"would silently drop them. Set taskType=TEXT_IMAGE to use them."
)
body: Final = _nova_canvas_task_body(
image_b64=image_b64,
@ -376,6 +490,9 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
task_type=task_type,
mask_prompt=mask_prompt,
out_painting_mode=out_painting_mode,
control_mode=control_mode,
control_strength=control_strength,
style=style,
)
# BACKGROUND_REMOVAL InvokeModel body must not include imageGenerationConfig (AWS rejects it).

View file

@ -30,7 +30,7 @@ from litellm.llms.custom_httpx.http_handler import (
from litellm.types.utils import ImageResponse
from ..base_aws_llm import BaseAWSLLM, bedrock_bearer_token
from ..common_utils import BedrockError
from ..common_utils import BedrockError, redact_bedrock_headers_for_logging
if TYPE_CHECKING:
from botocore.awsrequest import AWSPreparedRequest
@ -256,7 +256,10 @@ class BedrockImageEdit(BaseAWSLLM):
additional_args={
"complete_input_dict": data,
"api_base": proxy_endpoint_url,
"headers": prepped.headers,
# Redacted copy: pre_call forwards additional_args unmasked to
# logger_fn / log_pre_api_call callbacks; the signed headers
# must not leave the request path (prepped keeps them).
"headers": redact_bedrock_headers_for_logging(prepped.headers),
},
)
return BedrockImageEditPreparedRequest(

View file

@ -0,0 +1,8 @@
"""Bedrock video generation (Amazon Nova Reel via StartAsyncInvoke)."""
from typing import Final
from litellm.llms.bedrock.videos.handler import BedrockVideoGeneration
from litellm.llms.bedrock.videos.transformation import BedrockNovaReelVideoConfig
__all__: Final[tuple[str, ...]] = ("BedrockNovaReelVideoConfig", "BedrockVideoGeneration")

View file

@ -0,0 +1,118 @@
"""Bedrock dispatch shims for the main-layer video functions.
The generic video routes in ``litellm.videos.main`` delegate their bedrock
branches here (AWS SigV4 signing and the async-invoke API need the bedrock
handler instead of the shared HTTP handler). Each function lazy-imports
``BedrockVideoGeneration`` so importing this module never pulls the boto3
signing stack, and forwards its arguments verbatim.
"""
from __future__ import annotations
from collections.abc import Coroutine, Mapping
from typing import TYPE_CHECKING, Final
import httpx
from litellm.types.router import GenericLiteLLMParams
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.types.videos.main import VideoObject
def dispatch_bedrock_video_generation(
*,
model: str,
prompt: str,
video_generation_request_params: Mapping[str, object],
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLogging | None,
timeout: float | httpx.Timeout | None,
is_async: bool,
client: object | None = None,
extra_headers: dict[str, object] | None = None,
api_key: str | None = None,
) -> VideoObject | Coroutine[object, object, VideoObject]:
"""Create (StartAsyncInvoke) through the bedrock handler.
Merges the ``aws_*`` auth params riding on litellm_params into the optional
params for the handler (mirrors how images/main.py merges non_default_params
for bedrock) and threads the real litellm_params through so
metadata.request_id reaches the clientRequestToken fallback.
"""
from litellm.llms.bedrock.videos.handler import BedrockVideoGeneration
bedrock_optional_params: Final[dict[str, object]] = (
dict( # mutable-ok: aws_* params are merged in before the handler call
video_generation_request_params
)
)
bedrock_optional_params.update(
{ # mutable-ok: aws_* auth params merged into the bedrock params
k: v for k, v in litellm_params.model_dump(exclude_none=True).items() if k.startswith("aws_")
}
)
return BedrockVideoGeneration().video_generation(
model=model,
prompt=prompt,
optional_params=bedrock_optional_params,
logging_obj=logging_obj,
timeout=timeout,
avideo_generation=is_async,
client=client,
api_base=litellm_params.get("api_base"),
extra_headers=extra_headers,
api_key=api_key,
# Real litellm_params so metadata.request_id reaches the
# clientRequestToken fallback (idempotent retries); aws_* keys are
# already merged into bedrock_optional_params above and are never
# consumed from this object by the handler.
litellm_params=litellm_params,
)
def dispatch_bedrock_video_status(
*,
video_id: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLogging | None,
api_base: str | None,
api_key: str | None,
astatus: bool,
timeout: float | httpx.Timeout | None,
) -> VideoObject | Coroutine[object, object, VideoObject]:
"""Status (GetAsyncInvoke) through the bedrock handler."""
from litellm.llms.bedrock.videos.handler import BedrockVideoGeneration
return BedrockVideoGeneration().video_status(
video_id=video_id,
litellm_params=litellm_params,
logging_obj=logging_obj,
api_base=api_base,
api_key=api_key,
astatus=astatus,
timeout=timeout,
)
def dispatch_bedrock_video_content(
*,
video_id: str,
litellm_params: GenericLiteLLMParams,
logging_obj: LiteLLMLogging | None,
api_base: str | None,
api_key: str | None,
timeout: float | httpx.Timeout | None,
) -> bytes:
"""Content download (S3 output object) through the bedrock handler."""
from litellm.llms.bedrock.videos.handler import BedrockVideoGeneration
return BedrockVideoGeneration().video_content(
video_id=video_id,
litellm_params=litellm_params,
logging_obj=logging_obj,
api_base=api_base,
api_key=api_key,
timeout=timeout,
)

View file

@ -0,0 +1,773 @@
"""
Bedrock Nova Reel video handler.
Implements the LiteLLM video surface for ``amazon.nova-reel-v1:0`` (and regional
inference-profile ids) on top of the Bedrock asynchronous invoke API:
- create: POST {runtime}/async-invoke (StartAsyncInvoke, SigV4-signed)
- status: GET {runtime}/async-invoke/{arn} (GetAsyncInvoke, SigV4-signed)
- content: download ``output.mp4`` from the S3 output location via boto3
Mirrors the URL + signing conventions of
``litellm/llms/bedrock/embed/embedding.py`` (same async-invoke endpoints).
"""
from __future__ import annotations
import json
from collections.abc import Coroutine, Mapping, Sequence
from typing import TYPE_CHECKING, Final, TypeAlias
from urllib.parse import quote
import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.llms.bedrock.videos.transformation import BedrockNovaReelVideoConfig
from litellm.secret_managers.main import get_secret
from litellm.types.llms.bedrock import (
BedrockAsyncInvokeOutputDataConfig,
BedrockAsyncInvokeS3OutputDataConfig,
BedrockGetAsyncInvokeResponse,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.videos.main import VideoObject
from litellm.types.videos.utils import decode_video_id_with_provider, extract_original_video_id
from ..base_aws_llm import (
AWSPreparedRequest,
BaseAWSLLM,
Credentials,
bedrock_bearer_token,
pop_aws_auth_params,
)
from ..common_utils import BedrockError, BedrockModelInfo, redact_bedrock_headers_for_logging
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
_LitellmParamsDict: TypeAlias = dict[str, object]
_ExtraHeadersDict: TypeAlias = dict[str, object]
DEFAULT_VIDEO_REGION: Final = "us-west-2"
NOVA_REEL_OUTPUT_FILENAME: Final = "output.mp4"
# S3 download client defaults when no per-call timeout is given (video_content
# threads its call timeout in; these bound the boto3 client otherwise).
DEFAULT_S3_CONNECT_TIMEOUT_S: Final = 5.0
DEFAULT_S3_READ_TIMEOUT_S: Final = 60.0
def _client_error_code(err: Exception) -> str:
"""AWS error Code from a botocore ClientError response, '' when absent."""
response: Final = getattr(err, "response", None)
if isinstance(response, Mapping):
error: Final = response.get("Error")
if isinstance(error, Mapping):
code: Final = error.get("Code")
if isinstance(code, str):
return code
return ""
def _client_error_http_status(err: Exception) -> int | None:
"""HTTPStatusCode from a botocore ClientError ResponseMetadata, None when absent."""
response: Final = getattr(err, "response", None)
if isinstance(response, Mapping):
metadata: Final = response.get("ResponseMetadata")
if isinstance(metadata, Mapping):
status: Final = metadata.get("HTTPStatusCode")
if isinstance(status, int):
return status
return None
def _sign_get_request(
credentials: Credentials | None,
url: str,
headers: Mapping[str, str],
aws_region_name: str,
bearer_token: str | None = None,
) -> AWSPreparedRequest:
try:
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.exceptions import NoCredentialsError
except ImportError:
raise ImportError("Missing boto3 to call bedrock. Run 'pip install boto3'.")
request: Final = AWSRequest(
method="GET",
url=url,
data=None,
headers=(
{ # mutable-ok: merged headers dict carries the bearer token
**headers,
"Authorization": f"Bearer {bearer_token}",
}
if bearer_token is not None
else headers
),
)
if credentials is None and bearer_token is None:
# Fail fast the same way the shared POST signer does (base_aws_llm.py):
# an unsigned request would 403 at AWS with a far less actionable error.
raise NoCredentialsError()
if credentials is not None and bearer_token is None:
SigV4Auth(credentials, "bedrock", aws_region_name).add_auth(request)
return request.prepare()
def _region_from_invocation_arn(invocation_arn: str) -> str | None:
"""arn:aws:bedrock:{region}:{account}:async-invoke/{id} -> region."""
parts: Final = invocation_arn.split(":")
if len(parts) >= 4 and parts[0] == "arn":
return parts[3] or None
return None
def _parse_s3_uri(s3_uri: str) -> tuple[str, str]:
"""s3://bucket/optional/prefix/ -> (bucket, 'optional/prefix')."""
if not s3_uri.startswith("s3://"):
raise BedrockError(
status_code=400,
message=f"Invalid S3 output URI (expected s3://bucket/prefix): {s3_uri!r}",
)
trimmed: Final = s3_uri.rstrip("/")
without_scheme: Final = trimmed[len("s3://") :]
bucket, _, prefix = without_scheme.partition("/")
if not bucket:
raise BedrockError(status_code=400, message=f"Invalid S3 output URI: {s3_uri!r}")
return bucket, prefix
def _s3_uri_from_output_config(
output_config: BedrockAsyncInvokeOutputDataConfig | None,
) -> str | None:
"""s3Uri from the invocation's outputDataConfig.s3OutputDataConfig, if present."""
if output_config is None:
return None
s3_config: Final[BedrockAsyncInvokeS3OutputDataConfig | None] = output_config.get("s3OutputDataConfig")
if s3_config is None:
return None
s3_uri: Final[str | None] = s3_config.get("s3Uri")
return s3_uri
def _params_to_dict(litellm_params: GenericLiteLLMParams | Mapping[str, object] | None) -> _LitellmParamsDict:
"""GenericLiteLLMParams is a pydantic model with dict-like access; copy to a real dict."""
if litellm_params is None:
return {} # mutable-ok: empty params dict for None input
if isinstance(litellm_params, dict):
return dict(litellm_params) # mutable-ok: aws_* keys are popped in place downstream
if isinstance(litellm_params, GenericLiteLLMParams):
return litellm_params.model_dump(exclude_none=True)
return dict(litellm_params) # mutable-ok: aws_* keys are popped in place downstream
def _as_generic_litellm_params(
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None,
) -> GenericLiteLLMParams:
"""Normalize handler-level litellm_params to GenericLiteLLMParams for the create transform.
Only the clientRequestToken fallback inputs are read downstream
(metadata.request_id, then the litellm_call_id extra field); aws_* credential
keys are consumed from optional_params only, so nothing passed here
double-processes credentials.
"""
if isinstance(litellm_params, GenericLiteLLMParams):
return litellm_params
params: Final[GenericLiteLLMParams] = GenericLiteLLMParams()
if litellm_params is not None:
metadata: Final = litellm_params.get("metadata")
if isinstance(metadata, dict):
params.metadata = metadata # pyright: ignore[reportAttributeAccessIssue] # extra-allowed field
call_id: Final = litellm_params.get("litellm_call_id")
if isinstance(call_id, str) and call_id:
params.litellm_call_id = call_id # pyright: ignore[reportAttributeAccessIssue] # @client extra field
return params
class BedrockVideoGeneration(BaseAWSLLM):
"""
Bedrock video generation handler for Amazon Nova Reel models.
"""
def get_config_class(self) -> type[BedrockNovaReelVideoConfig]:
return BedrockNovaReelVideoConfig
def _load_credentials(
self,
optional_params: dict, # mutable-ok: aws_* keys are popped in place
aws_region_name: str | None = None,
bearer_token: str | None = None,
) -> tuple[Credentials | None, str]:
"""Resolve SigV4 credentials + region the same way BedrockEmbedding does."""
auth_params: Final = pop_aws_auth_params(optional_params)
if aws_region_name is None:
aws_region_name = optional_params.pop("aws_region_name", None)
if aws_region_name is None:
litellm_aws_region_name: Final = get_secret("AWS_REGION_NAME", None)
if litellm_aws_region_name is not None and isinstance(litellm_aws_region_name, str):
aws_region_name = litellm_aws_region_name
standard_aws_region_name: Final = get_secret("AWS_REGION", None)
if standard_aws_region_name is not None and isinstance(standard_aws_region_name, str):
aws_region_name = standard_aws_region_name
if aws_region_name is None:
aws_region_name = DEFAULT_VIDEO_REGION
credentials: Final[Credentials | None] = (
None if bearer_token is not None else self.resolve_credentials(auth_params, aws_region_name)
)
return credentials, aws_region_name
def _prepare_async_invoke_request(
self,
model: str,
prompt: str,
optional_params: _LitellmParamsDict,
api_base: str | None,
extra_headers: _ExtraHeadersDict | None,
logging_obj: LiteLLMLogging | None,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None = None,
) -> tuple[str, AWSPreparedRequest, bytes, _LitellmParamsDict]:
"""
Returns (endpoint_url, prepped_request, body, data) for POST /async-invoke.
litellm_params feeds only the clientRequestToken fallback
(metadata.request_id); aws_* credentials are consumed from
optional_params, never from litellm_params, so nothing is double-processed.
"""
bearer_token: Final = bedrock_bearer_token(api_key)
boto3_credentials_info: Final = self._get_boto_credentials_from_optional_params(
optional_params, model, bearer_token=bearer_token
)
bedrock_provider: Final = self.get_bedrock_invoke_provider(model)
model_id: Final = self.get_bedrock_model_id(
model=model,
provider=bedrock_provider,
optional_params=optional_params,
)
_, proxy_endpoint_url = self.get_runtime_endpoint(
api_base=api_base,
aws_bedrock_runtime_endpoint=boto3_credentials_info.aws_bedrock_runtime_endpoint,
aws_region_name=boto3_credentials_info.aws_region_name,
)
endpoint_url: Final = f"{proxy_endpoint_url.rstrip('/')}/async-invoke"
config: Final = BedrockNovaReelVideoConfig()
data, _, _ = config.transform_video_create_request(
model=model_id,
prompt=prompt,
api_base=endpoint_url,
video_create_optional_request_params=optional_params,
litellm_params=_as_generic_litellm_params(litellm_params),
headers={}, # mutable-ok: transform never reads headers for Nova Reel
)
# The transform returns the model name it was given; make sure the
# envelope carries the resolved Bedrock model id.
data["modelId"] = model_id
body: Final = json.dumps(data).encode("utf-8")
headers: Final[_ExtraHeadersDict] = {
"Content-Type": "application/json",
**(extra_headers or {}),
}
prepped: Final = self.get_request_headers(
credentials=boto3_credentials_info.credentials,
aws_region_name=boto3_credentials_info.aws_region_name,
extra_headers=extra_headers,
endpoint_url=endpoint_url,
data=body,
headers=headers,
api_key=api_key,
)
if logging_obj is not None:
logging_obj.pre_call(
input=prompt,
api_key="",
additional_args={ # mutable-ok: logging payload dict built for this call
"complete_input_dict": data,
"api_base": endpoint_url,
# Redacted copy: pre_call forwards additional_args unmasked to
# logger_fn / log_pre_api_call callbacks; the signed headers
# must not leave the request path (prepped keeps them).
"headers": redact_bedrock_headers_for_logging(prepped.headers),
},
)
return endpoint_url, prepped, body, data
def _transform_create_response(
self,
model: str,
response: httpx.Response,
data: Mapping[str, object],
logging_obj: LiteLLMLogging | None,
) -> VideoObject:
if logging_obj is not None:
logging_obj.post_call(
input="",
api_key="",
original_response=response.text,
additional_args={"complete_input_dict": data}, # mutable-ok: logging payload dict built for this call
)
# raise_for_status() already ran on both call paths, so any non-2xx is
# covered; manual status_code checks here would only misclassify
# legitimate 2xx variants (e.g. 202 from proxies) behind raise_for_status.
config: Final = BedrockNovaReelVideoConfig()
return config.transform_video_create_response(
model=model,
raw_response=response,
logging_obj=logging_obj,
request_data=data,
)
def video_generation(
self,
model: str,
prompt: str,
optional_params: _LitellmParamsDict,
logging_obj: LiteLLMLogging | None,
timeout: float | httpx.Timeout | None,
avideo_generation: bool = False,
client: httpx.Client | httpx.AsyncClient | None = None,
api_base: str | None = None,
extra_headers: _ExtraHeadersDict | None = None,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None = None,
) -> VideoObject | Coroutine[object, object, VideoObject]:
"""Returns a VideoObject, or a coroutine resolving to one when avideo_generation is set."""
if avideo_generation:
return self.async_video_generation(
model=model,
prompt=prompt,
optional_params=optional_params,
logging_obj=logging_obj,
timeout=timeout,
client=client,
api_base=api_base,
extra_headers=extra_headers,
api_key=api_key,
litellm_params=litellm_params,
)
endpoint_url, prepped, body, data = self._prepare_async_invoke_request(
model=model,
prompt=prompt,
optional_params=optional_params,
api_base=api_base,
extra_headers=extra_headers,
logging_obj=logging_obj,
api_key=api_key,
litellm_params=litellm_params,
)
from litellm.llms.custom_httpx.http_handler import _get_httpx_client
sync_client: Final = client if isinstance(client, httpx.Client) else _get_httpx_client()
try:
response: Final = sync_client.post(
url=endpoint_url,
headers=prepped.headers,
content=body,
timeout=timeout,
)
response.raise_for_status()
except httpx.HTTPStatusError as err:
raise BedrockError(
status_code=err.response.status_code,
message=err.response.text,
headers=err.response.headers,
response=err.response,
)
except httpx.TimeoutException:
raise BedrockError(status_code=408, message="Timeout error occurred.")
return self._transform_create_response(model, response, data, logging_obj)
async def async_video_generation(
self,
model: str,
prompt: str,
optional_params: _LitellmParamsDict,
logging_obj: LiteLLMLogging | None,
timeout: float | httpx.Timeout | None,
client: httpx.Client | httpx.AsyncClient | None = None,
api_base: str | None = None,
extra_headers: _ExtraHeadersDict | None = None,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None = None,
) -> VideoObject:
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
endpoint_url, prepped, body, data = self._prepare_async_invoke_request(
model=model,
prompt=prompt,
optional_params=optional_params,
api_base=api_base,
extra_headers=extra_headers,
logging_obj=logging_obj,
api_key=api_key,
litellm_params=litellm_params,
)
async_client: Final = (
client
if isinstance(client, httpx.AsyncClient)
else get_async_httpx_client(
llm_provider=litellm.LlmProviders.BEDROCK,
params={"timeout": timeout}, # mutable-ok: per-call timeout kwargs for the shared client factory
)
)
try:
response: Final = await async_client.post(
url=endpoint_url,
headers=prepped.headers,
content=body,
timeout=timeout,
)
response.raise_for_status()
except httpx.HTTPStatusError as err:
raise BedrockError(
status_code=err.response.status_code,
message=err.response.text,
headers=err.response.headers,
response=err.response,
)
except httpx.TimeoutException:
raise BedrockError(status_code=408, message="Timeout error occurred.")
return self._transform_create_response(model, response, data, logging_obj)
def _status_request_parts(
self,
invocation_arn: str,
optional_params: dict, # mutable-ok: aws_* keys are popped in place
api_base: str | None,
api_key: str | None = None,
) -> tuple[str, AWSPreparedRequest, str]:
"""Returns (status_url, prepped_get_request, resolved_region) for GET /async-invoke/{arn}.
The resolved region (explicit aws_region_name > ARN region > env > default) is
returned so callers that follow up with an S3 download reuse the same region
instead of re-resolving to the env/default one.
"""
bearer_token: Final = bedrock_bearer_token(api_key)
aws_region_name: str | None = optional_params.pop("aws_region_name", None)
if aws_region_name is None:
aws_region_name = _region_from_invocation_arn(invocation_arn)
credentials, aws_region_name = self._load_credentials(
optional_params,
aws_region_name=aws_region_name,
bearer_token=bearer_token,
)
_, proxy_endpoint_url = self.get_runtime_endpoint(
api_base=api_base,
aws_bedrock_runtime_endpoint=optional_params.pop("aws_bedrock_runtime_endpoint", None),
aws_region_name=aws_region_name,
)
encoded_arn: Final = quote(invocation_arn, safe="")
status_url: Final = f"{proxy_endpoint_url.rstrip('/')}/async-invoke/{encoded_arn}"
prepped: Final = _sign_get_request(
credentials=credentials,
url=status_url,
headers={"Content-Type": "application/json"}, # mutable-ok: SigV4 signs a plain headers dict
aws_region_name=aws_region_name,
bearer_token=bearer_token,
)
return status_url, prepped, aws_region_name
def _decode_status_context(self, video_id: str) -> tuple[str, str]:
"""Returns (invocation_arn, model) encoded in the video id (single decode)."""
decoded: Final = decode_video_id_with_provider(video_id)
invocation_arn: Final[str] = decoded.get("video_id") or extract_original_video_id(video_id)
if not invocation_arn:
raise BedrockError(
status_code=400,
message=f"Could not extract a Bedrock invocation ARN from video id: {video_id!r}",
)
model: Final[str] = decoded.get("model_id") or "amazon.nova-reel-v1:0"
return invocation_arn, model
def _sync_get(self, prepped: AWSPreparedRequest, timeout: float | httpx.Timeout | None = None) -> httpx.Response:
from litellm.llms.custom_httpx.http_handler import _get_httpx_client
client: Final = _get_httpx_client()
try:
return client.get(url=prepped.url, headers=prepped.headers, timeout=timeout)
except httpx.TimeoutException:
raise BedrockError(status_code=408, message="Timeout error occurred.")
async def _async_get(
self,
prepped: AWSPreparedRequest,
timeout: float | httpx.Timeout | None = None,
) -> httpx.Response:
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
client: Final = get_async_httpx_client(
llm_provider=litellm.LlmProviders.BEDROCK,
params={"timeout": timeout}, # mutable-ok: per-call timeout kwargs for the shared client factory
)
try:
return await client.get(url=prepped.url, headers=prepped.headers, timeout=timeout)
except httpx.TimeoutException:
raise BedrockError(status_code=408, message="Timeout error occurred.")
def _map_status_response(
self,
response: httpx.Response,
model: str,
video_id: str,
logging_obj: LiteLLMLogging | None,
) -> tuple[VideoObject, BedrockGetAsyncInvokeResponse]:
if response.status_code != 200:
raise BedrockError(
status_code=response.status_code,
message=f"Nova Reel get-async-invoke error: {response.text}",
headers=response.headers,
response=response,
)
try:
# Guarded parse: the transform's own non-JSON guard never runs because
# raw is needed here first; a non-JSON 200 must map to a 502, not a 500.
raw: BedrockGetAsyncInvokeResponse = response.json()
except ValueError as err:
raise BedrockError(
status_code=502,
message=f"non-JSON response from Bedrock status endpoint (expected GetAsyncInvoke JSON): {err}",
) from err
config: Final = BedrockNovaReelVideoConfig()
video_obj = config.transform_video_status_retrieve_response(
raw_response=response,
logging_obj=logging_obj,
model=model,
video_id=video_id,
)
return video_obj, raw
def video_status(
self,
video_id: str,
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None = None,
logging_obj: LiteLLMLogging | None = None,
api_base: str | None = None,
api_key: str | None = None,
astatus: bool = False,
timeout: float | httpx.Timeout | None = None,
) -> VideoObject | Coroutine[object, object, VideoObject]:
"""Returns a VideoObject, or a coroutine resolving to one when astatus is set."""
invocation_arn, model = self._decode_status_context(video_id)
optional_params: Final[_LitellmParamsDict] = _params_to_dict(litellm_params)
_, prepped, _ = self._status_request_parts(invocation_arn, optional_params, api_base, api_key=api_key)
if astatus:
return self._async_video_status(
prepped=prepped, model=model, video_id=video_id, logging_obj=logging_obj, timeout=timeout
)
response: Final = self._sync_get(prepped, timeout=timeout)
video_obj, _ = self._map_status_response(response, model, video_id, logging_obj)
return video_obj
async def _async_video_status(
self,
prepped: AWSPreparedRequest,
model: str,
video_id: str,
logging_obj: LiteLLMLogging | None = None,
timeout: float | httpx.Timeout | None = None,
) -> VideoObject:
"""Private async arm of video_status; only its internal dispatch calls it."""
response: Final = await self._async_get(prepped, timeout=timeout)
video_obj, _ = self._map_status_response(response, model, video_id, logging_obj)
return video_obj
def video_content(
self,
video_id: str,
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None = None,
logging_obj: LiteLLMLogging | None = None,
api_base: str | None = None,
api_key: str | None = None,
timeout: float | httpx.Timeout | None = None,
) -> bytes:
"""Download output.mp4 from the S3 output location once the job completed."""
invocation_arn, model = self._decode_status_context(video_id)
optional_params: Final[_LitellmParamsDict] = _params_to_dict(litellm_params)
_, prepped, status_region = self._status_request_parts(
invocation_arn, optional_params, api_base, api_key=api_key
)
response: Final = self._sync_get(prepped, timeout=timeout)
video_obj, raw = self._map_status_response(response, model, video_id, logging_obj)
if video_obj.status == "failed":
failure_message: Final[str] = (
(video_obj.error.get("message") if video_obj.error else None) or raw.get("failureMessage") or ""
)
raise BedrockError(
status_code=502,
message=f"Nova Reel invocation failed: {failure_message}",
)
if video_obj.status != "completed":
# Client error about job state: 400-class, not a 500 that SDKs retry.
raise BedrockError(
status_code=400,
message=(
"Nova Reel video generation is not complete yet "
f"(status={video_obj.status}). Check video_status() before downloading."
),
)
s3_uri: Final[str | None] = _s3_uri_from_output_config(raw.get("outputDataConfig"))
if not s3_uri:
# The raw get-async-invoke payload carries invocationArn (with the
# account id); debug-log it, raise with only the observed key names.
verbose_logger.debug("Nova Reel completed invocation without an S3 output location: %r", raw)
output_config: Final = raw.get("outputDataConfig")
observed_keys: Final = tuple(sorted(output_config.keys())) if isinstance(output_config, Mapping) else ()
raise ValueError(
f"No S3 output location on completed invocation (observed outputDataConfig keys: {observed_keys})"
)
bucket, prefix = _parse_s3_uri(s3_uri)
# Nova Reel v1:1 writes output.mp4 into a per-invocation folder under the
# configured prefix (AWS docs); only the older v1:0 flows placed it flat
# under the prefix. With a shared prefix the flat key can hold a foreign
# or stale object, so the flat fallback is v1:0-only. The modelArn on the
# get-async-invoke response is authoritative over the id-encoded model
# (inference profiles can encode a base id while the ARN names the
# actually-invoked variant).
model_arn: Final[object] = raw.get("modelArn")
resolved_model: Final[str] = model_arn.rsplit("/", 1)[-1] if isinstance(model_arn, str) and model_arn else model
gate_model: Final[str] = resolved_model if "nova-reel" in resolved_model else model
allow_flat: Final[bool] = BedrockModelInfo.get_base_model(gate_model) == "amazon.nova-reel-v1:0"
key_candidates: Final[list[str]] = [ # mutable-ok: candidate S3 keys are tried in order
f"{prefix}/{invocation_arn.rsplit('/', 1)[-1]}/{NOVA_REEL_OUTPUT_FILENAME}".lstrip("/"),
*(
[f"{prefix}/{NOVA_REEL_OUTPUT_FILENAME}".lstrip("/")] # mutable-ok: v1:0-only flat fallback key
if allow_flat
else []
),
]
return self._download_s3_object(
bucket,
key_candidates,
litellm_params,
raw,
region_default=status_region,
api_key=api_key,
timeout=timeout,
)
def _download_s3_object(
self,
bucket: str,
key_candidates: Sequence[str],
litellm_params: GenericLiteLLMParams | Mapping[str, object] | None,
raw: BedrockGetAsyncInvokeResponse,
region_default: str | None = None,
api_key: str | None = None,
timeout: float | httpx.Timeout | None = None,
) -> bytes:
"""Download the output object from S3.
region_default (the region the status request resolved: explicit
aws_region_name > ARN region > env > default) is used unless the fresh
litellm_params carry an explicit aws_region_name, which still wins.
timeout (the video_content call timeout) bounds the boto3 client's
read timeout; connect timeout is it or DEFAULT_S3_CONNECT_TIMEOUT_S,
whichever is smaller.
"""
try:
import boto3
from botocore.config import Config as BotocoreConfig
from botocore.exceptions import BotoCoreError, ClientError, NoCredentialsError
except ImportError:
raise ImportError("Missing boto3 to download Nova Reel output. Run 'pip install boto3'.")
optional_params: Final[_LitellmParamsDict] = _params_to_dict(litellm_params)
explicit_region: Final[str | None] = optional_params.pop("aws_region_name", None)
bearer_token: Final[str | None] = bedrock_bearer_token(api_key)
bearer_s3_guidance: Final = (
"Nova Reel video content download requires AWS SigV4 credentials with S3 read "
"access (aws_access_key_id/aws_secret_access_key or an ambient credential chain). "
"Bedrock bearer tokens only cover the Bedrock asynchronous invoke API and cannot "
"download objects from S3."
)
# Always resolve SigV4 credentials for S3: a Bedrock bearer token only
# covers the async-invoke API, never S3 object access. With a bearer token
# in play and no resolvable SigV4 credentials, resolve_credentials raises
# NoCredentialsError before the guidance check below can fire; catch it so
# the bearer guidance 400 (not the generic mapping) reaches the caller.
try:
credentials, region = self._load_credentials(
optional_params,
aws_region_name=(explicit_region if explicit_region is not None else region_default),
)
except NoCredentialsError as err:
if bearer_token is not None:
raise BedrockError(status_code=400, message=bearer_s3_guidance) from err
raise # no bearer in play: unchanged propagation
if bearer_token is not None and credentials is None:
raise BedrockError(
status_code=400,
message=bearer_s3_guidance,
)
session_kwargs: Final[dict[str, str]] = {"region_name": region} # mutable-ok: credential keys are added below
if credentials is not None:
session_kwargs["aws_access_key_id"] = credentials.access_key
session_kwargs["aws_secret_access_key"] = credentials.secret_key
if credentials.token:
session_kwargs["aws_session_token"] = credentials.token
read_timeout: Final[float] = float(timeout) if isinstance(timeout, (int, float)) else DEFAULT_S3_READ_TIMEOUT_S
connect_timeout: Final[float] = min(DEFAULT_S3_CONNECT_TIMEOUT_S, read_timeout)
client_config: Final = BotocoreConfig(connect_timeout=connect_timeout, read_timeout=read_timeout)
session: Final = boto3.Session(**session_kwargs)
s3_client: Final = session.client("s3", config=client_config)
s3_uri: Final[str | None] = _s3_uri_from_output_config(raw.get("outputDataConfig"))
errors: list[str] = [] # mutable-ok: error strings accumulate across candidate keys
try:
for key in key_candidates:
body = None # per-iteration resource; closed in finally
try:
obj = s3_client.get_object(Bucket=bucket, Key=key)
body = obj["Body"]
return body.read()
except ClientError as err:
# Annotated, not Final: basedpyright forbids Final assignment inside loops.
error_code: str = _client_error_code(err)
if error_code == "NoSuchKey" or _client_error_http_status(err) == 404:
# Only a missing key falls through to the next candidate.
errors.append(str(err))
continue
if error_code == "AccessDenied":
raise BedrockError(
status_code=403,
message=(f"Access denied downloading Nova Reel output from {s3_uri} (key {key!r}): {err}"),
) from err
raise BedrockError(
status_code=502,
message=(
f"AWS error (code {error_code!r}) downloading Nova Reel output "
f"from {s3_uri} (key {key!r}): {err}"
),
) from err
finally:
if body is not None:
body.close()
except BotoCoreError as err:
# ClientError never reaches this handler: it inherits Exception (not
# BotoCoreError) and is handled per-key above. Anything else that is
# a BotoCoreError (NoCredentialsError, EndpointConnectionError, ...)
# aborts the download and maps to a 502. botocore messages carry
# class name + failure reason, never credentials.
raise BedrockError(
status_code=502,
message=f"Failed to download Nova Reel output from S3: {type(err).__name__}: {err}",
)
finally:
s3_client.close()
raise BedrockError(
status_code=404,
message=(
"Nova Reel output video not found in the S3 output location "
f"{s3_uri}. Tried keys: {tuple(key_candidates)}. Errors: {tuple(errors)}"
),
headers={}, # mutable-ok: synthesized 404 carries no provider headers
)

View file

@ -0,0 +1,765 @@
"""
Amazon Nova Reel video generation on Bedrock (StartAsyncInvoke / GetAsyncInvoke).
Nova Reel is invoked through the Bedrock asynchronous invoke API:
- create: POST {runtime}/async-invoke body: {modelId, modelInput, outputDataConfig}
- status: GET {runtime}/async-invoke/{arn} response: {invocationArn, status, failureMessage, ...}
- content: download output.mp4 from the S3 output location when status is Completed
The real HTTP (AWS SigV4 signing + S3 download) happens in
``litellm.llms.bedrock.videos.handler.BedrockVideoGeneration``; this config
builds/validates the request bodies and maps responses to ``VideoObject``.
Refs:
- https://docs.aws.amazon.com/nova/latest/userguide/video-req-resp-structure.html
- https://docs.aws.amazon.com/nova/latest/userguide/video-gen-access.html
- bedrock-runtime service model: StartAsyncInvoke POST /async-invoke,
GetAsyncInvoke GET /async-invoke/{invocationArn}, AsyncInvokeStatus enum
InProgress | Completed | Failed.
"""
from __future__ import annotations
import base64
import binascii
import re
from collections.abc import Mapping
from datetime import datetime
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, TypeAlias
import httpx
from httpx._types import FileContent, RequestFiles
from litellm._logging import verbose_logger
from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.types.llms.bedrock import (
BedrockGetAsyncInvokeResponse,
BedrockStartAsyncInvokeResponse,
)
from litellm.types.router import GenericLiteLLMParams
from litellm.types.videos.main import VideoCreateOptionalRequestParams
from litellm.types.videos.utils import (
decode_video_id_with_provider,
encode_video_id_with_provider,
extract_original_video_id,
)
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.types.videos.main import CharacterObject, VideoObject
_SupportedParams: TypeAlias = list[str]
_VideoParams: TypeAlias = dict[str, object]
_VideoHeaders: TypeAlias = dict[str, str]
_VideoStringParams: TypeAlias = dict[str, str]
NOVA_REEL_DEFAULT_DURATION_SECONDS: Final = 6
NOVA_REEL_DEFAULT_FPS: Final = 24
NOVA_REEL_DEFAULT_DIMENSION: Final = "1280x720"
# AWS clientRequestToken: alphanumeric and hyphens only, at most 64 chars.
NOVA_REEL_CLIENT_REQUEST_TOKEN_MAX_LEN: Final = 64
_NOVA_REEL_TOKEN_UNSAFE: Final = re.compile(r"[^0-9A-Za-z-]")
# AWS async-invoke status enum (bedrock-runtime service model) -> OpenAI-style
# VideoObject.status values used across LiteLLM video providers.
NOVA_REEL_STATUS_MAP: Final[Mapping[str, str]] = MappingProxyType(
{
"InProgress": "processing",
"Completed": "completed",
"Failed": "failed",
}
)
_UNSUPPORTED_MESSAGE: Final = (
"video {operation} is not supported for Bedrock Nova Reel; Nova Reel exposes "
"create (video_generation), status (video_status) and content (video_content) only"
)
def _unsupported_operation_error(operation: str) -> BedrockError:
"""400-class error for unsupported video operations.
Verified against litellm.exception_type: a plain ValueError/NotImplementedError
both fall through to APIConnectionError (500-class), while BedrockError carries
status_code=400 into BadRequestError through the bedrock mapping.
"""
return BedrockError(status_code=400, message=_UNSUPPORTED_MESSAGE.format(operation=operation))
def _user_input_error(message: str) -> BedrockError:
"""400-class error for invalid caller input (same mapping as _unsupported_operation_error).
A plain ValueError would surface as APIConnectionError (500-class, which OpenAI
SDKs auto-retry); BedrockError(status_code=400) maps to BadRequestError instead.
"""
return BedrockError(status_code=400, message=message)
def _data_url_payload(image: str) -> str:
"""Validate a ``data:<mime>;base64,`` URL and return only its payload."""
header, sep, encoded = image.partition(",")
if not sep or not header.removeprefix("data:").endswith(";base64"):
raise _user_input_error(
"Nova Reel input_reference data URLs must be base64-encoded "
"(data:image/png;base64,<payload>); got an unsupported data URL prefix."
)
return encoded
def _file_content_to_b64_and_format(image: FileContent) -> tuple[str, str]:
"""Read an input-reference image and return (base64, "png"|"jpeg")."""
if isinstance(image, bytes):
image_bytes: bytes = image
elif isinstance(image, str):
# Base64-encoded string, optionally wrapped in a data URL; detect the
# format from the decoded header.
payload: Final = _data_url_payload(image) if image.startswith("data:") else image
try:
image_bytes = base64.b64decode(payload, validate=True)
except (binascii.Error, ValueError) as err:
raise _user_input_error(f"Nova Reel input_reference string did not decode as base64: {err}") from err
elif hasattr(image, "read") and callable(getattr(image, "read", None)):
if hasattr(image, "seek"):
image.seek(0)
image_bytes = image.read()
if not isinstance(image_bytes, bytes):
raise _user_input_error(
"Nova Reel input_reference file objects must be opened in binary mode "
f"(read() returned {type(image_bytes).__name__}); open image files with 'rb'."
)
else:
raise _user_input_error(
f"Nova Reel input_reference must be bytes, a file-like object or a base64 string; got {type(image)!r}"
)
if image_bytes.startswith(b"\x89PNG"):
image_format: str = "png"
elif image_bytes.startswith(b"\xff\xd8"):
image_format = "jpeg"
else:
raise _user_input_error(
"Nova Reel input_reference images must be PNG or JPEG encoded; "
f"unrecognized image header bytes {image_bytes[:8]!r}"
)
return base64.b64encode(image_bytes).decode("utf-8"), image_format
def _duration_seconds_from_request(request_data: Mapping[str, object] | None) -> float | None:
"""durationSeconds from the StartAsyncInvoke request envelope, for cost calculation.
TEXT_VIDEO and MULTI_SHOT_AUTOMATED carry a single durationSeconds on
videoGenerationConfig; MULTI_SHOT_MANUAL carries per-shot durations inside
multiShotManualParams.shots[*].durationSeconds, so the billable duration is
the sum of the shot durations.
"""
if request_data is None:
return None
model_input: Final[object | None] = request_data.get("modelInput")
if not isinstance(model_input, Mapping):
return None
manual_params: Final[object | None] = model_input.get("multiShotManualParams")
if isinstance(manual_params, Mapping):
shots: Final[object | None] = manual_params.get("shots")
if isinstance(shots, list):
total = 0.0
saw_duration = False
for shot in shots:
if not isinstance(shot, Mapping):
continue
# Annotated, not Final: basedpyright forbids Final assignment inside loops.
shot_duration: object | None = shot.get("durationSeconds")
if not isinstance(shot_duration, (int, float, str)):
continue
try:
total += float(shot_duration)
saw_duration = True
except ValueError:
continue
if saw_duration:
return total
generation_config: Final[object | None] = model_input.get("videoGenerationConfig")
if not isinstance(generation_config, Mapping):
return None
duration: Final[object | None] = generation_config.get("durationSeconds")
if not isinstance(duration, (int, float, str)):
return None
try:
return float(duration)
except ValueError:
return None
def _sanitize_client_request_token(token: str) -> str:
"""AWS clientRequestToken allows alphanumeric and hyphens, max 64 chars."""
return _NOVA_REEL_TOKEN_UNSAFE.sub("-", token)[:NOVA_REEL_CLIENT_REQUEST_TOKEN_MAX_LEN]
def _request_id_from_litellm_params(litellm_params: GenericLiteLLMParams) -> str | None:
"""Best-effort litellm request id: metadata.request_id, then the litellm_call_id extra field."""
metadata: Final = getattr(litellm_params, "metadata", None)
if isinstance(metadata, Mapping):
request_id: Final = metadata.get("request_id")
if isinstance(request_id, str) and request_id:
return request_id
call_id: Final = getattr(litellm_params, "litellm_call_id", None)
if isinstance(call_id, str) and call_id:
return call_id
return None
def _generation_config_from_op(op: _VideoParams, task_type: object) -> _VideoParams:
"""videoGenerationConfig from request params (pops seconds/size/dimension/fps/seed).
durationSeconds lives on videoGenerationConfig for TEXT_VIDEO and
MULTI_SHOT_AUTOMATED only; MULTI_SHOT_MANUAL durations live per shot
inside multiShotManualParams.shots, so it is omitted there.
"""
generation_config: Final[_VideoParams] = {
"fps": NOVA_REEL_DEFAULT_FPS,
"dimension": NOVA_REEL_DEFAULT_DIMENSION,
}
single_duration: Final[bool] = task_type != "MULTI_SHOT_MANUAL"
if single_duration:
generation_config["durationSeconds"] = NOVA_REEL_DEFAULT_DURATION_SECONDS
seconds: Final = op.pop("seconds", None)
if isinstance(seconds, (int, float, str)):
try:
parsed_seconds: Final = int(float(seconds))
except ValueError as err:
raise _user_input_error(f"Nova Reel seconds must be a number; got {seconds!r}") from err
if single_duration:
generation_config["durationSeconds"] = parsed_seconds
size: Final = op.pop("size", None)
if size is not None and isinstance(size, str) and "x" in size:
generation_config["dimension"] = size.replace(" ", "")
dimension: Final = op.pop("dimension", None)
if dimension is not None and isinstance(dimension, str) and dimension.strip():
generation_config["dimension"] = dimension
fps: Final = op.pop("fps", None)
if fps is not None:
try:
generation_config["fps"] = int(float(fps))
except ValueError as err:
raise _user_input_error(f"Nova Reel fps must be a number; got {fps!r}") from err
seed: Final = op.pop("seed", None)
if seed is not None:
try:
generation_config["seed"] = int(float(seed))
except ValueError as err:
raise _user_input_error(f"Nova Reel seed must be a number; got {seed!r}") from err
return generation_config
def _task_params_from_op(task_type: object, op: _VideoParams, prompt: str, input_reference: object) -> _VideoParams:
"""Per-taskType params section for modelInput (pops multiShot params from op)."""
if task_type == "MULTI_SHOT_AUTOMATED":
# AWS schema: automated multi-shot takes multiShotAutomatedParams
# (never textToVideoParams) and forbids input images.
if input_reference is not None:
raise _user_input_error(
"Nova Reel MULTI_SHOT_AUTOMATED does not accept input images "
"(input_reference/image); automated multi-shot is text-driven only."
)
automated_params: Final[object | None] = op.pop("multiShotAutomatedParams", None)
automated_section: Final[_VideoParams] = {
"multiShotAutomatedParams": (
automated_params if isinstance(automated_params, Mapping) else {"text": prompt}
)
}
return automated_section
if task_type == "MULTI_SHOT_MANUAL":
manual_params: Final[object | None] = op.pop("multiShotManualParams", None)
if not isinstance(manual_params, Mapping):
raise _user_input_error(
"Nova Reel MULTI_SHOT_MANUAL requires multiShotManualParams in the request "
"(shot definitions with per-shot text/images/durationSeconds); "
f"got {manual_params!r}."
)
manual_section: Final[_VideoParams] = {"multiShotManualParams": manual_params}
return manual_section
# TEXT_VIDEO (default): textToVideoParams with the prompt and optional images.
text_to_video_params: Final[_VideoParams] = {"text": prompt}
if input_reference is not None:
if isinstance(input_reference, dict):
# Pre-built provider shape: {"format": ..., "source": {...}}
text_to_video_params["images"] = [input_reference] # mutable-ok: AWS images param is a list
else:
image_b64, image_format = _file_content_to_b64_and_format(
input_reference # pyright: ignore[reportArgumentType] # untyped user input; helper validates
)
text_to_video_params["images"] = [ # mutable-ok: AWS images param is a list
{"format": image_format, "source": {"bytes": image_b64}} # mutable-ok: nested AWS image payload
]
text_section: Final[_VideoParams] = {"textToVideoParams": text_to_video_params}
return text_section
class BedrockNovaReelVideoConfig(BaseVideoConfig):
"""
Video config for amazon.nova-reel-v1:0 (and regional variants) on Bedrock.
Health checks: the proxy's video_generation probe calls avideo_generation()
with only a prompt, which this config rejects with a 400 because Nova Reel
requires a per-request output_s3_uri; the deployment is then reported
unhealthy (the health check marks any errored probe unhealthy, 4xx included).
Every video provider's probe creates a real video, so there is no cheaper
repo-consistent probe. Operators should set
``model_info.disable_background_health_check: true`` on Nova Reel
deployments (or put ``output_s3_uri`` in the deployment litellm_params and
accept that each health check starts a real, billed generation).
"""
def get_supported_openai_params(self, model: str) -> _SupportedParams:
return [ # mutable-ok: BaseVideoConfig requires a list
"seconds",
"size",
"seed",
"fps",
"dimension",
"taskType",
"output_s3_uri",
"kmsKeyId",
"bucketOwner",
"input_reference",
"image",
]
def map_openai_params(
self,
video_create_optional_params: VideoCreateOptionalRequestParams,
model: str,
drop_params: bool,
) -> _VideoParams:
# All supported params pass through untouched; Nova Reel-specific keys
# keep their AWS names (durationSeconds etc. are built in
# transform_video_create_request from seconds/size).
return dict(video_create_optional_params) # mutable-ok: BaseVideoConfig requires a mutable param mapping
def validate_environment(
self,
headers: _VideoHeaders,
model: str,
api_key: str | None = None,
litellm_params: GenericLiteLLMParams | None = None,
) -> _VideoHeaders:
if headers is None:
headers = {} # mutable-ok: None headers start empty before Content-Type is added
if "Content-Type" not in headers:
headers["Content-Type"] = "application/json"
return headers
def get_error_class(
self,
error_message: str,
status_code: int,
headers: dict | httpx.Headers, # mutable-ok: BaseVideoConfig passes headers as a dict
) -> BedrockError:
"""BedrockError synthesizes a response that keeps provider headers like x-amzn-RequestId."""
return BedrockError(status_code=status_code, message=error_message, headers=headers)
def get_complete_url(
self,
model: str,
api_base: str | None,
litellm_params: _VideoParams,
) -> str:
# The shared OpenAI-style video handlers resolve the URL before running
# any config transform, so the transform-level unsupported-operation
# guards never run for those routes. Nova Reel create/status/content
# bypass these handlers entirely (litellm.llms.bedrock.videos.dispatch
# builds the signed requests), so get_complete_url is only reachable
# for unsupported operations: raise the same 400-class BedrockError the
# transforms raise instead of a NotImplementedError that would surface
# as a 500-class APIConnectionError.
raise BedrockError(
status_code=400,
message=(
"bedrock video supports create, status and content only; this "
"request reached the shared video handler, which only happens "
"for unsupported operations (edit, characters, remix, extension, "
"list, delete)"
),
)
def transform_video_create_request(
self,
model: str,
prompt: str,
api_base: str,
video_create_optional_request_params: _VideoParams,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
) -> tuple[_VideoParams, RequestFiles, str]:
"""
Build the StartAsyncInvoke request body.
Returns (request_body, files, "POST") where request_body is the wrapped
async-invoke envelope {modelId, modelInput, outputDataConfig}.
"""
op: Final[_VideoParams] = dict( # mutable-ok: request params are popped in place while building modelInput
video_create_optional_request_params
)
output_s3_uri: Final = op.pop("output_s3_uri", None)
if not output_s3_uri or not str(output_s3_uri).strip():
raise _user_input_error(
"Nova Reel video generation requires an S3 output location. Pass "
'output_s3_uri="s3://my-bucket/optional-prefix/" in the request '
"(Bedrock writes output.mp4 there)."
)
if not str(output_s3_uri).startswith("s3://"):
raise _user_input_error(
"Nova Reel output_s3_uri must be an S3 URI starting with s3:// "
f"(got {output_s3_uri!r}); Bedrock writes output.mp4 into that bucket."
)
# Pop the S3 config keys (and their snake_case aliases) before the
# modelInput passthrough merge so they never leak into modelInput; they
# are forwarded into outputDataConfig.s3OutputDataConfig below.
kms_key_id: Final = op.pop("kmsKeyId", None) or op.pop("output_s3_kms_key_id", None)
bucket_owner: Final = op.pop("bucketOwner", None) or op.pop("output_s3_bucket_owner", None)
task_type: Final = op.pop("taskType", "TEXT_VIDEO")
if not str(prompt or "").strip():
raise _user_input_error("Nova Reel prompt is required and cannot be empty (or whitespace-only).")
# Pop both reference keys unconditionally so neither leaks into modelInput;
# input_reference wins when a caller passes both.
popped_reference: Final = op.pop("input_reference", None)
popped_image: Final = op.pop("image", None)
input_reference: Final = popped_reference if popped_reference is not None else popped_image
model_input: Final[_VideoParams] = {
"taskType": task_type,
"videoGenerationConfig": _generation_config_from_op(op, task_type),
}
model_input.update(_task_params_from_op(task_type, op, prompt, input_reference))
# Known non-AWS video-client params have no Nova Reel mapping; drop them
# instead of leaking junk keys into modelInput. Everything else keeps the
# verbatim passthrough (provider-specific AWS keys like multiShotManualParams).
for dropped_key in ("parameters", "resolution", "characters", "user", "extra_headers"):
op.pop(dropped_key, None)
# Pop both token keys unconditionally (caller-supplied wins over the
# litellm request id) so neither leaks into modelInput.
caller_token_snake: Final = op.pop("client_request_token", None)
caller_token_camel: Final = op.pop("clientRequestToken", None)
caller_token: Final = caller_token_snake if caller_token_snake is not None else caller_token_camel
model_input.update(op)
request_id: Final[str | None] = _request_id_from_litellm_params(litellm_params)
client_request_token: Final[str | None] = (
_sanitize_client_request_token(str(caller_token))
if caller_token is not None
else (_sanitize_client_request_token(request_id) if request_id is not None else None)
)
# Optional S3 config keys are added below before the envelope is returned.
s3_output_config: Final[_VideoParams] = {"s3Uri": output_s3_uri}
if kms_key_id is not None:
s3_output_config["kmsKeyId"] = kms_key_id
if bucket_owner is not None:
s3_output_config["bucketOwner"] = bucket_owner
request_body: Final[_VideoParams] = {
"modelId": model,
"modelInput": model_input,
"outputDataConfig": {"s3OutputDataConfig": s3_output_config},
}
if client_request_token:
# Envelope key only when populated: caller-supplied token or litellm request id.
request_body["clientRequestToken"] = client_request_token
return request_body, [], "POST" # mutable-ok: HTTP files payload requires a list
def transform_video_create_response(
self,
model: str,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
custom_llm_provider: str | None = None,
request_data: Mapping[str, object] | None = None,
) -> VideoObject:
from litellm.types.videos.main import VideoObject
try:
response_data: Final[BedrockStartAsyncInvokeResponse] = raw_response.json()
except ValueError as err:
raise BedrockError(
status_code=502,
message=f"Nova Reel async-invoke returned a non-JSON response: {err}",
) from err
invocation_arn: Final[str | None] = response_data.get("invocationArn")
if not invocation_arn:
raise ValueError(f"Nova Reel async-invoke response missing invocationArn: {response_data}")
video_obj = VideoObject(
id=encode_video_id_with_provider(invocation_arn, "bedrock", model),
object="video",
status="processing",
model=model,
created_at=_epoch_now(),
)
duration_seconds: Final[float | None] = _duration_seconds_from_request(request_data)
if duration_seconds is not None:
# Mirrors the Vertex video config: lets the video cost calculator
# compute cost from output_cost_per_second * duration_seconds.
video_obj.usage = {"duration_seconds": duration_seconds} # mutable-ok: VideoObject.usage payload dict
return video_obj
def transform_video_status_retrieve_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
) -> tuple[str, _VideoParams]:
raise NotImplementedError(
"Nova Reel status URLs are built and signed in BedrockVideoGeneration "
"(GET /async-invoke/{arn}). Do not use this transform for this config."
)
def transform_video_status_retrieve_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
custom_llm_provider: str | None = None,
client: HTTPHandler | None = None,
model: str | None = None,
video_id: str | None = None,
) -> VideoObject:
from litellm.types.videos.main import VideoObject
try:
response_data: Final[BedrockGetAsyncInvokeResponse] = raw_response.json()
except ValueError as err:
raise BedrockError(
status_code=502,
message=f"Nova Reel get-async-invoke returned a non-JSON response: {err}",
) from err
invocation_arn: Final[str | None] = response_data.get("invocationArn")
if not invocation_arn:
raise ValueError(f"Nova Reel get-async-invoke response missing invocationArn: {response_data}")
status_field: Final[object] = response_data.get("status")
if not isinstance(status_field, str) or not status_field:
raise BedrockError(
status_code=500,
message=(
"Nova Reel get-async-invoke response had an unexpected shape: "
f"missing or empty 'status' (observed keys: {sorted(response_data.keys())})"
),
)
raw_status: Final[str] = status_field
if raw_status not in NOVA_REEL_STATUS_MAP:
verbose_logger.warning("Nova Reel unmapped invocationStatus=%r; reporting processing", raw_status)
status: Final[str] = NOVA_REEL_STATUS_MAP.get(raw_status, "processing")
failure_message: Final[str | None] = response_data.get("failureMessage")
video_obj = VideoObject(
id=encode_video_id_with_provider(invocation_arn, "bedrock", model),
object="video",
status=status,
model=model,
created_at=_to_epoch(response_data.get("submitTime")),
completed_at=(_to_epoch(response_data.get("endTime")) if status == "completed" else None),
error=(
{"message": failure_message} # mutable-ok: VideoObject.error accepts a plain payload dict
if status == "failed" and failure_message
else None
),
)
output_config: Final = response_data.get("outputDataConfig")
if output_config is not None:
s3_config: Final = output_config.get("s3OutputDataConfig")
if s3_config is not None:
s3_uri: Final[str | None] = s3_config.get("s3Uri")
if s3_uri:
# Provider detail, not usage: rides on _hidden_params (like the
# vertex video transforms' provider-specific fields) so cost
# calculators reading usage.duration_seconds never trip on it.
video_obj._hidden_params["output_s3_uri"] = s3_uri
return video_obj
def transform_video_content_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
variant: str | None = None,
) -> tuple[str, _VideoParams]:
raise NotImplementedError(
"Nova Reel video content is downloaded from the S3 output location in "
"BedrockVideoGeneration. Do not use this transform for this config."
)
def transform_video_content_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
) -> bytes:
return raw_response.content
def transform_video_remix_request(
self,
video_id: str,
prompt: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
extra_body: Mapping[str, object] | None = None,
) -> tuple[str, _VideoParams]:
raise _unsupported_operation_error("remix")
def transform_video_remix_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
custom_llm_provider: str | None = None,
) -> VideoObject:
raise _unsupported_operation_error("remix")
def transform_video_list_request(
self,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
after: str | None = None,
limit: int | None = None,
order: str | None = None,
extra_query: Mapping[str, object] | None = None,
) -> tuple[str, _VideoParams]:
raise _unsupported_operation_error("list")
def transform_video_list_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
custom_llm_provider: str | None = None,
) -> _VideoStringParams:
raise _unsupported_operation_error("list")
def transform_video_delete_request(
self,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
) -> tuple[str, _VideoParams]:
raise _unsupported_operation_error("delete")
def transform_video_delete_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
) -> VideoObject:
raise _unsupported_operation_error("delete")
def transform_video_create_character_request(
self,
name: str,
video: object, # base declares Any; never read, this override always raises
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
) -> tuple[str, list]:
raise _unsupported_operation_error("create character")
def transform_video_create_character_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
) -> CharacterObject:
raise _unsupported_operation_error("create character")
def transform_video_get_character_request(
self,
character_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
) -> tuple[str, _VideoParams]:
raise _unsupported_operation_error("get character")
def transform_video_get_character_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
) -> CharacterObject:
raise _unsupported_operation_error("get character")
def transform_video_edit_request(
self,
prompt: str,
video_id: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
video_file: FileContent | None = None,
extra_body: Mapping[str, object] | None = None,
prefetched_source_data: dict[str, object] | None = None,
) -> tuple[str, Mapping[str, object], RequestFiles | None]:
raise _unsupported_operation_error("edit")
def transform_video_edit_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
custom_llm_provider: str | None = None,
request_data: dict | None = None,
) -> VideoObject:
raise _unsupported_operation_error("edit")
def transform_video_extension_request(
self,
prompt: str,
video_id: str,
seconds: str,
api_base: str,
litellm_params: GenericLiteLLMParams,
headers: _VideoHeaders,
extra_body: Mapping[str, object] | None = None,
) -> tuple[str, _VideoParams]:
raise _unsupported_operation_error("extension")
def transform_video_extension_response(
self,
raw_response: httpx.Response,
logging_obj: LiteLLMLogging | None,
custom_llm_provider: str | None = None,
) -> VideoObject:
raise _unsupported_operation_error("extension")
@staticmethod
def extract_invocation_arn(video_id: str) -> str:
"""Return the raw Bedrock invocationArn from a (possibly encoded) video id."""
decoded: Final = decode_video_id_with_provider(video_id)
arn: Final = decoded.get("video_id") or ""
return arn or extract_original_video_id(video_id)
def _epoch_now() -> int:
import time
return int(time.time())
def _to_epoch(timestamp: str | float | None) -> int | None:
"""Bedrock timestamps to unix epoch seconds.
The bedrock-runtime Smithy model declares timestampFormat: iso8601 for
submitTime/endTime, so real GetAsyncInvoke payloads carry strings like
"2026-01-15T10:30:00Z"; numeric epochs are accepted too.
"""
if timestamp is None:
return None
try:
return int(float(timestamp))
except (TypeError, ValueError):
pass
try:
return int(datetime.fromisoformat(str(timestamp).replace("Z", "+00:00")).timestamp())
except ValueError:
verbose_logger.warning("Nova Reel response carried an unparseable timestamp %r; leaving it unset", timestamp)
return None

View file

@ -292,6 +292,42 @@
"output_cost_per_image": 0.06,
"supports_nova_canvas_image_edit": true
},
"amazon.nova-reel-v1:0": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"amazon.nova-reel-v1:1": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"us.amazon.nova-reel-v1:0": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"us.amazon.nova-reel-v1:1": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"us.writer.palmyra-x4-v1:0": {
"input_cost_per_token": 2.5e-06,
"litellm_provider": "bedrock_converse",

View file

@ -250,7 +250,8 @@
"vector_stores_search": true,
"count_tokens": true,
"rag_ingest": true,
"rag_query": true
"rag_query": true,
"video_generations": true
}
},
"s3_vectors": {

View file

@ -1,6 +1,6 @@
#### Video Endpoints #####
from typing import Final
from typing import TYPE_CHECKING, Final
from fastapi import APIRouter, Depends, File, Form, Request, Response, UploadFile
from fastapi.responses import ORJSONResponse
@ -27,9 +27,36 @@ from litellm.types.videos.utils import (
decode_video_id_with_provider,
)
if TYPE_CHECKING:
from litellm.router import Router
router: Final = APIRouter()
def _resolve_model_name_from_decoded_model_id(
llm_router: "Router",
model_id_from_decoded: str,
custom_llm_provider: str | None,
) -> str | None:
"""Resolve the router model_name from the model id encoded in a video/character id.
Bedrock cross-region inference-profile ids (e.g. ``us.amazon.nova-reel-v1:1``)
do not match deployments configured with the base model id
(``bedrock/amazon.nova-reel-v1:1``), which would drop the deployment's
litellm_params (aws_* credentials) on status/content calls. Mirrors the Nova
Canvas image-edit transform: strip the region prefix via
``BedrockModelInfo.get_base_model`` and retry the resolution (bedrock only).
"""
resolved_model = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
if resolved_model is None and custom_llm_provider == "bedrock":
from litellm.llms.bedrock.common_utils import BedrockModelInfo
base_model: Final = BedrockModelInfo.get_base_model(model_id_from_decoded)
if base_model != model_id_from_decoded:
resolved_model = llm_router.resolve_model_name_from_model_id(base_model)
return resolved_model
@router.post(
"/v1/videos",
dependencies=[Depends(user_api_key_auth)],
@ -269,7 +296,9 @@ async def video_status(
# Resolve model_name from model_id if available
# This allows the router to automatically inject litellm_params from the model config
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
resolved_model: Final = _resolve_model_name_from_decoded_model_id(
llm_router, model_id_from_decoded, custom_llm_provider
)
if resolved_model:
data["model"] = resolved_model
@ -369,7 +398,9 @@ async def video_content(
# Resolve model_name from model_id if available
# This allows the router to automatically inject litellm_params from the model config
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
resolved_model: Final = _resolve_model_name_from_decoded_model_id(
llm_router, model_id_from_decoded, custom_llm_provider
)
if resolved_model:
data["model"] = resolved_model
# Process request using ProxyBaseLLMRequestProcessing
@ -477,7 +508,9 @@ async def video_remix(
# Resolve model_name from model_id if available
# This allows the router to automatically inject litellm_params from the model config
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
resolved_model: Final = _resolve_model_name_from_decoded_model_id(
llm_router, model_id_from_decoded, custom_llm_provider
)
if resolved_model:
data["model"] = resolved_model
@ -680,7 +713,9 @@ async def video_get_character(
data["custom_llm_provider"] = custom_llm_provider
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
resolved_model: Final = _resolve_model_name_from_decoded_model_id(
llm_router, model_id_from_decoded, custom_llm_provider
)
if resolved_model:
data["model"] = resolved_model
@ -791,7 +826,9 @@ async def video_edit(
data["custom_llm_provider"] = custom_llm_provider
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
resolved_model: Final = _resolve_model_name_from_decoded_model_id(
llm_router, model_id_from_decoded, custom_llm_provider
)
if resolved_model:
data["model"] = resolved_model
@ -888,7 +925,9 @@ async def video_extension(
data["custom_llm_provider"] = custom_llm_provider
if model_id_from_decoded and llm_router:
resolved_model: Final = llm_router.resolve_model_name_from_model_id(model_id_from_decoded)
resolved_model: Final = _resolve_model_name_from_decoded_model_id(
llm_router, model_id_from_decoded, custom_llm_provider
)
if resolved_model:
data["model"] = resolved_model

View file

@ -914,13 +914,29 @@ class AmazonNovaCanvasImageGenerationConfig(TypedDict, total=False):
class AmazonNovaCanvasTextToImageParams(TypedDict, total=False):
"""
Params for Amazon Nova Canvas Text to Image API
conditionImage + controlMode + controlStrength enable conditioned editing
(SEGMENTATION derives a segmentation mask from the condition image;
CANNY_EDGE follows its prominent contours and is the AWS default).
"""
text: str
negativeText: str
controlStrength: float
controlMode: Literal["CANNY_EDIT", "SEGMENTATION"]
controlMode: ReadOnly[Literal["CANNY_EDGE", "SEGMENTATION"]]
conditionImage: str
style: ReadOnly[
Literal[
"3D_ANIMATED_FAMILY_FILM",
"DESIGN_SKETCH",
"FLAT_VECTOR_ILLUSTRATION",
"GRAPHIC_NOVEL_ILLUSTRATION",
"MAXIMALISM",
"MIDCENTURY_RETRO",
"PHOTOREALISM",
"SOFT_DIGITAL_PAINTING",
]
]
class AmazonNovaCanvasTextToImageRequest(AmazonNovaCanvasRequestBase, TypedDict, total=False):
@ -1001,6 +1017,175 @@ class AmazonTitanImageGenerationRequestBody(TypedDict, total=False):
imageGenerationConfig: AmazonNovaCanvasImageGenerationConfig
################ Amazon Nova Reel Video Types ################
NOVA_REEL_TASK_TYPES = Literal["TEXT_VIDEO", "MULTI_SHOT_AUTOMATED", "MULTI_SHOT_MANUAL"]
class AmazonNovaReelS3Location(TypedDict, total=False):
"""
S3 location for a Nova Reel input image.
Ref: https://docs.aws.amazon.com/nova/latest/userguide/video-req-resp-structure.html
"""
uri: ReadOnly[str]
bucketOwner: ReadOnly[str]
class AmazonNovaReelImageSourceLocation(TypedDict, total=False):
"""
Location of a Nova Reel input image: inline base64 bytes or S3.
"""
bytes: ReadOnly[str] # base64 encoded image
s3Location: ReadOnly[AmazonNovaReelS3Location]
class AmazonNovaReelImageSource(TypedDict, total=False):
"""
Image source for Nova Reel textToVideoParams.images entries.
"""
format: ReadOnly[Literal["png", "jpeg"]]
source: ReadOnly[AmazonNovaReelImageSourceLocation]
class AmazonNovaReelTextToVideoParams(TypedDict, total=False):
"""
Params for Amazon Nova Reel text/image-to-video generation.
Ref: https://docs.aws.amazon.com/nova/latest/userguide/video-req-resp-structure.html
"""
text: ReadOnly[str]
images: ReadOnly[Sequence[AmazonNovaReelImageSource]]
class AmazonNovaReelVideoGenerationConfig(TypedDict, total=False):
"""
Generation config for Amazon Nova Reel.
durationSeconds: 6 for single-shot (v1:0 supports 6|10; v1:1 single-shot is 6);
multiples of 6 up to 120 for multi-shot. fps: 24 only. dimension: "1280x720"
(v1:0 also supports "720x1280"). seed: 0-2147483646, AWS default 42.
"""
durationSeconds: ReadOnly[int]
fps: ReadOnly[int]
dimension: ReadOnly[str]
seed: ReadOnly[int]
class AmazonNovaReelMultiShotAutomatedParams(TypedDict, total=False):
"""
Params for Nova Reel MULTI_SHOT_AUTOMATED: text-driven shot planning only
(no input images; durationSeconds stays on videoGenerationConfig).
"""
text: ReadOnly[str]
class AmazonNovaReelMultiShotManualShot(TypedDict, total=False):
"""
One shot of MULTI_SHOT_MANUAL: per-shot text, optional input images and
durationSeconds (durations live per shot, not on videoGenerationConfig).
"""
text: ReadOnly[str]
durationSeconds: ReadOnly[int]
images: ReadOnly[Sequence[AmazonNovaReelImageSource]]
class AmazonNovaReelMultiShotManualParams(TypedDict, total=False):
"""
Params for Nova Reel MULTI_SHOT_MANUAL (required for that task type).
"""
shots: ReadOnly[Sequence[AmazonNovaReelMultiShotManualShot]]
class AmazonNovaReelModelInput(TypedDict, total=False):
"""
modelInput body for Nova Reel StartAsyncInvoke.
TEXT_VIDEO (default) uses textToVideoParams; MULTI_SHOT_AUTOMATED uses
multiShotAutomatedParams (no input images); MULTI_SHOT_MANUAL requires
multiShotManualParams and omits top-level videoGenerationConfig.durationSeconds.
"""
taskType: ReadOnly[NOVA_REEL_TASK_TYPES]
textToVideoParams: ReadOnly[AmazonNovaReelTextToVideoParams]
multiShotAutomatedParams: ReadOnly[AmazonNovaReelMultiShotAutomatedParams]
multiShotManualParams: ReadOnly[AmazonNovaReelMultiShotManualParams]
videoGenerationConfig: ReadOnly[AmazonNovaReelVideoGenerationConfig]
class BedrockAsyncInvokeS3OutputDataConfig(TypedDict, total=False):
"""
S3 output config for Bedrock StartAsyncInvoke (nested outputDataConfig key).
Ref: bedrock-runtime service model (StartAsyncInvokeRequest.outputDataConfig)
"""
s3Uri: ReadOnly[str]
kmsKeyId: ReadOnly[str]
bucketOwner: ReadOnly[str]
class BedrockAsyncInvokeOutputDataConfig(TypedDict, total=False):
"""
Output data config for Bedrock StartAsyncInvoke.
"""
s3OutputDataConfig: ReadOnly[BedrockAsyncInvokeS3OutputDataConfig]
class BedrockStartAsyncInvokeRequest(TypedDict, total=False):
"""
Request body for POST {runtime}/async-invoke (StartAsyncInvoke).
Ref: https://docs.aws.amazon.com/nova/latest/userguide/video-gen-access.html
"""
modelId: ReadOnly[str]
modelInput: ReadOnly[AmazonNovaReelModelInput]
outputDataConfig: ReadOnly[BedrockAsyncInvokeOutputDataConfig]
clientRequestToken: ReadOnly[str]
class BedrockStartAsyncInvokeResponse(TypedDict, total=False):
"""
Response body for POST {runtime}/async-invoke.
"""
invocationArn: ReadOnly[str]
BEDROCK_ASYNC_INVOKE_STATUSES = Literal["InProgress", "Completed", "Failed"]
class BedrockGetAsyncInvokeResponse(TypedDict, total=False):
"""
Response body for GET {runtime}/async-invoke/{invocationArn} (GetAsyncInvoke).
status enum verified against the bedrock-runtime service model
(AsyncInvokeStatus): InProgress | Completed | Failed. failureMessage is
present when status is Failed. Timestamps use the model's iso8601
timestampFormat ("2026-01-15T10:30:00Z"); numeric epochs are tolerated.
"""
invocationArn: ReadOnly[str]
modelArn: ReadOnly[str]
clientRequestToken: ReadOnly[str]
status: ReadOnly[BEDROCK_ASYNC_INVOKE_STATUSES]
failureMessage: ReadOnly[str]
submitTime: ReadOnly[str | float]
lastModifiedTime: ReadOnly[str | float]
endTime: ReadOnly[str | float]
outputDataConfig: ReadOnly[BedrockAsyncInvokeOutputDataConfig]
if TYPE_CHECKING:
from botocore.awsrequest import AWSPreparedRequest
else:

View file

@ -9669,6 +9669,15 @@ class ProviderConfigManager:
from litellm.llms.hosted_vllm.videos import get_hosted_vllm_video_config
return get_hosted_vllm_video_config(model)
elif LlmProviders.BEDROCK == provider:
from litellm.llms.bedrock.videos.transformation import BedrockNovaReelVideoConfig
# model is None for status/content routes (the provider config only
# does response mapping there; the real model id is encoded in the
# video id itself).
if model is None or "nova-reel" in model.lower():
return BedrockNovaReelVideoConfig()
return None
elif LlmProviders.EDENAI == provider:
return litellm.EdenAIVideoConfig()
return None

View file

@ -13,6 +13,11 @@ from litellm.constants import request_timeout as DEFAULT_REQUEST_TIMEOUT
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
from litellm.llms.bedrock.videos.dispatch import (
dispatch_bedrock_video_content,
dispatch_bedrock_video_generation,
dispatch_bedrock_video_status,
)
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.main import base_llm_http_handler
from litellm.types.router import GenericLiteLLMParams
@ -243,6 +248,21 @@ def video_generation(
# Set the correct call type for video generation
litellm_logging_obj.call_type = CallTypes.create_video.value
# Route bedrock to its specific handler (AWS SigV4 signing required)
if custom_llm_provider == "bedrock":
return dispatch_bedrock_video_generation(
model=model,
prompt=prompt,
video_generation_request_params=video_generation_request_params,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
timeout=timeout or DEFAULT_REQUEST_TIMEOUT,
is_async=_is_async,
client=kwargs.get("client"),
extra_headers=extra_headers,
api_key=kwargs.get("api_key") or litellm_params.get("api_key"),
)
# Call the handler with _is_async flag instead of directly calling the async handler
return base_llm_http_handler.video_generation_handler(
model=model,
@ -355,6 +375,15 @@ def video_content(
)
# Call the handler with _is_async flag instead of directly calling the async handler
if custom_llm_provider == "bedrock":
return dispatch_bedrock_video_content(
video_id=video_id,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
api_base=litellm_params.get("api_base"),
api_key=kwargs.get("api_key") or litellm_params.get("api_key"),
timeout=timeout or DEFAULT_REQUEST_TIMEOUT,
)
return base_llm_http_handler.video_content_handler(
video_id=video_id,
video_content_provider_config=video_provider_config,
@ -1058,6 +1087,18 @@ def video_status(
# Set the correct call type for video status
litellm_logging_obj.call_type = CallTypes.video_retrieve.value
# Route bedrock to its specific handler (AWS SigV4 signing required)
if custom_llm_provider == "bedrock":
return dispatch_bedrock_video_status(
video_id=video_id,
litellm_params=litellm_params,
logging_obj=litellm_logging_obj,
api_base=litellm_params.get("api_base"),
api_key=kwargs.get("api_key") or litellm_params.get("api_key"),
astatus=_is_async,
timeout=timeout or DEFAULT_REQUEST_TIMEOUT,
)
# Call the handler with _is_async flag instead of directly calling the async handler
return base_llm_http_handler.video_status_handler(
video_id=video_id,

View file

@ -292,6 +292,42 @@
"output_cost_per_image": 0.06,
"supports_nova_canvas_image_edit": true
},
"amazon.nova-reel-v1:0": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"amazon.nova-reel-v1:1": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"us.amazon.nova-reel-v1:0": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"us.amazon.nova-reel-v1:1": {
"litellm_provider": "bedrock",
"mode": "video_generation",
"output_cost_per_second": 0.08,
"source": "https://aws.amazon.com/bedrock/pricing/ (AWS Bulk Price List API, AmazonBedrock offer 2026-09-22) (accessed 2026-09-24)",
"supported_endpoints": ["/v1/videos"],
"supported_modalities": ["text", "image"],
"supported_output_modalities": ["video"]
},
"us.writer.palmyra-x4-v1:0": {
"input_cost_per_token": 2.5e-06,
"litellm_provider": "bedrock_converse",

View file

@ -268,7 +268,8 @@
"vector_stores_search": true,
"count_tokens": true,
"rag_ingest": true,
"rag_query": true
"rag_query": true,
"video_generations": true
}
},
"s3_vectors": {

View file

@ -0,0 +1,795 @@
"""Tests for the Nova Canvas conditioned-editing additions (issue #39552).
Self-contained complement to the branch-added tests in
tests/unit/llms/bedrock/image_edit/test_amazon_nova_canvas_image_edit.py: the
conditioning guard, conditionImage acceptance/precedence, controlStrength
coercion/range, mask/maskPrompt rejection, prompt-required, style forwarding,
supported params, and the litellm-level proxy contracts (400-class validation
errors, style passthrough through litellm.aimage_edit).
"""
import asyncio
import base64
import io
import json
from datetime import datetime
from typing import Final
from unittest.mock import patch
import httpx
import pytest
import litellm
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.image_edit.amazon_nova_canvas_image_edit_transformation import (
BedrockAmazonNovaCanvasImageEditConfig,
)
TEST_MODEL = "amazon.nova-canvas-v1:0"
@pytest.fixture(autouse=True)
def ensure_nova_canvas_image_edit_model_cost_flags(monkeypatch):
"""Routing uses ``supports_nova_canvas_image_edit`` on ``litellm.model_cost``.
Full ``model_prices_and_context_window.json`` includes these flags, but CI or
alternate cost maps may omit them; merge minimal entries so tests match production.
"""
from litellm.utils import _invalidate_model_cost_lowercase_map
for key in (
"amazon.nova-canvas-v1:0",
"us.amazon.nova-canvas-v1:0",
):
entry = litellm.model_cost.get(key) or {}
if entry.get("supports_nova_canvas_image_edit") is True:
continue
monkeypatch.setitem(
litellm.model_cost,
key,
{
**entry,
"litellm_provider": entry.get("litellm_provider", "bedrock"),
"mode": entry.get("mode", "image_generation"),
"supports_nova_canvas_image_edit": True,
},
)
_invalidate_model_cost_lowercase_map()
yield
_invalidate_model_cost_lowercase_map()
#################################################
# conditioning guard: controlMode/controlStrength/style need TEXT_IMAGE
#################################################
def test_transform_request_conditioning_fields_without_text_image_task_type_raises():
"""controlMode/controlStrength/style with any non-TEXT_IMAGE resolved taskType must
fail fast instead of being silently dropped by mask/variation routing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"controlMode": "SEGMENTATION",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "taskType=TEXT_IMAGE" in str(excinfo.value.message)
assert "controlMode/controlStrength/style" in str(excinfo.value.message)
def test_transform_request_control_strength_without_text_image_task_type_raises():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"controlStrength": 0.4,
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "taskType=TEXT_IMAGE" in str(excinfo.value.message)
def test_transform_request_style_without_text_image_task_type_raises():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"style": "DESIGN_SKETCH",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "taskType=TEXT_IMAGE" in str(excinfo.value.message)
def test_transform_request_conditioning_fields_with_explicit_text_image_works():
"""The guard must not fire when taskType is explicitly TEXT_IMAGE."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "CANNY_EDGE",
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
assert body["textToImageParams"]["controlMode"] == "CANNY_EDGE"
#################################################
# conditionImage acceptance and precedence
#################################################
def test_transform_request_condition_image_without_multipart_image():
"""conditionImage is an alternative TEXT_IMAGE condition source for JSON-only callers."""
config = BedrockAmazonNovaCanvasImageEditConfig()
condition_b64: Final = base64.b64encode(b"cond-bytes").decode("utf-8")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="same layout",
image=None,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"conditionImage": condition_b64,
"controlMode": "SEGMENTATION",
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
t2i = body["textToImageParams"]
assert t2i["conditionImage"] == condition_b64
assert t2i["controlMode"] == "SEGMENTATION"
def test_transform_request_condition_image_bytes_accepted():
"""Raw bytes conditionImage is base64-encoded on the way into the body."""
config = BedrockAmazonNovaCanvasImageEditConfig()
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="same layout",
image=None,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"conditionImage": b"raw-cond-bytes",
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["conditionImage"] == base64.b64encode(b"raw-cond-bytes").decode("utf-8")
def test_transform_request_multipart_image_wins_over_condition_image():
"""Pinned precedence: when both are supplied the multipart `image` field wins."""
config = BedrockAmazonNovaCanvasImageEditConfig()
condition_b64: Final = base64.b64encode(b"from-condition-param").decode("utf-8")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="same layout",
image=io.BytesIO(b"from-multipart"),
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"conditionImage": condition_b64,
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["conditionImage"] == base64.b64encode(b"from-multipart").decode("utf-8")
def test_transform_request_condition_image_with_other_task_type_raises():
"""conditionImage only conditions TEXT_IMAGE; other task types need the image input."""
config = BedrockAmazonNovaCanvasImageEditConfig()
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=None,
image_edit_optional_request_params={
"taskType": "IMAGE_VARIATION",
"conditionImage": base64.b64encode(b"cond").decode("utf-8"),
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "conditionImage is only supported" in str(excinfo.value.message)
def test_transform_request_multipart_image_with_condition_image_non_text_image_raises():
"""image + conditionImage + a non-TEXT_IMAGE taskType must raise 400 instead of
silently discarding the conditionImage (the multipart image no longer short-circuits
the conflict check)."""
config = BedrockAmazonNovaCanvasImageEditConfig()
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=io.BytesIO(b"img-bytes"),
image_edit_optional_request_params={
"taskType": "IMAGE_VARIATION",
"conditionImage": base64.b64encode(b"cond").decode("utf-8"),
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "conditionImage is only supported" in str(excinfo.value.message)
def test_transform_request_inpainting_without_mask_maps_to_400():
"""INPAINTING without maskPrompt or maskImage must surface as a 400-class
BedrockError, not a plain ValueError (500-class through the proxy)."""
config = BedrockAmazonNovaCanvasImageEditConfig()
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="fix it",
image=io.BytesIO(b"img-bytes"),
image_edit_optional_request_params={"taskType": "INPAINTING"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "INPAINTING requires either maskPrompt or maskImage" in str(excinfo.value.message)
def test_transform_request_unsupported_task_type_maps_to_400():
config = BedrockAmazonNovaCanvasImageEditConfig()
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="x",
image=io.BytesIO(b"img-bytes"),
image_edit_optional_request_params={"taskType": "NOT_A_REAL_TASK_TYPE"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "Unsupported Amazon Nova Canvas taskType" in str(excinfo.value.message)
def test_transform_request_text_image_without_any_image_raises():
"""TEXT_IMAGE with neither a multipart image nor a conditionImage fails fast."""
config = BedrockAmazonNovaCanvasImageEditConfig()
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="same layout",
image=None,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "requires an image input" in str(excinfo.value.message)
def test_transform_request_text_image_invalid_control_mode_raises():
"""Unknown controlMode fails fast instead of hitting AWS with a bad payload."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE", "controlMode": "CANNY_EDIT"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "Unsupported Amazon Nova Canvas controlMode" in str(excinfo.value.message)
def test_transform_request_text_image_forwards_negative_text():
"""TEXT_IMAGE forwards negativeText into textToImageParams."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE", "negativeText": "blurry"},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["negativeText"] == "blurry"
#################################################
# controlStrength coercion and range
#################################################
def test_transform_request_text_image_control_strength_string_coerced():
"""controlStrength arriving as a string (multipart form data) is coerced to a float."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": "0.7",
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["controlStrength"] == 0.7
@pytest.mark.parametrize("control_strength", [0.0, 1.0])
def test_transform_request_text_image_control_strength_bounds_pass(control_strength):
"""Boundary controlStrength values 0.0 and 1.0 are accepted."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": control_strength,
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["controlStrength"] == control_strength
def test_transform_request_text_image_control_strength_out_of_range_raises():
"""controlStrength outside 0.0-1.0 fails fast."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": 1.5,
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "controlStrength must be between 0.0 and 1.0" in str(excinfo.value.message)
def test_transform_request_text_image_control_strength_non_numeric_string_raises():
"""A non-numeric controlStrength string must fail fast, not TypeError."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": "abc",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "controlStrength must be a number" in str(excinfo.value.message)
#################################################
# TEXT_IMAGE rejects mask inputs and requires a prompt
#################################################
def test_transform_request_text_image_with_mask_raises():
"""TEXT_IMAGE has no mask field; a provided mask must fail fast."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
mask = io.BytesIO(b"mask-bytes")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"mask": mask,
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "does not support a mask" in str(excinfo.value.message)
assert "INPAINTING or OUTPAINTING" in str(excinfo.value.message)
def test_transform_request_text_image_with_mask_prompt_raises():
"""TEXT_IMAGE has no maskPrompt field either; fail fast like the binary mask
instead of silently dropping it."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"maskPrompt": "the sky region",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "does not support a mask" in str(excinfo.value.message)
assert "INPAINTING or OUTPAINTING" in str(excinfo.value.message)
def test_transform_request_text_image_empty_prompt_raises():
"""An empty TEXT_IMAGE prompt must fail fast, not silently send a blank."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt="",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "TEXT_IMAGE requires a text prompt" in str(excinfo.value.message)
def test_transform_request_text_image_none_prompt_raises():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model=TEST_MODEL,
prompt=None,
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "TEXT_IMAGE requires a text prompt" in str(excinfo.value.message)
#################################################
# style forwarding and supported params
#################################################
def test_transform_request_text_image_forwards_style():
"""style is forwarded into textToImageParams for conditioned editing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="a city street in the same layout",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "SEGMENTATION",
"style": "DESIGN_SKETCH",
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
assert body["textToImageParams"]["style"] == "DESIGN_SKETCH"
def test_transform_request_text_image_omits_style_when_absent():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model=TEST_MODEL,
prompt="restyle",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert "style" not in body["textToImageParams"]
def test_get_supported_openai_params_includes_conditioning_fields():
"""controlMode/controlStrength are advertised for TEXT_IMAGE conditioned editing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
supported = config.get_supported_openai_params(TEST_MODEL)
assert "controlMode" in supported
assert "controlStrength" in supported
def test_get_supported_openai_params_includes_style():
"""style is advertised for TEXT_IMAGE conditioned editing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
supported = config.get_supported_openai_params(TEST_MODEL)
assert "style" in supported
def test_get_supported_openai_params_includes_condition_image():
config = BedrockAmazonNovaCanvasImageEditConfig()
supported = config.get_supported_openai_params(TEST_MODEL)
assert "conditionImage" in supported
#################################################
# litellm-level proxy contracts
#################################################
async def test_aimage_edit_mask_with_text_image_maps_to_bad_request(monkeypatch):
"""Through the litellm image-edit layer, the TEXT_IMAGE mask guard must surface as
litellm.BadRequestError (400-class), never APIConnectionError/500."""
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "env-bearer-token-12345")
with pytest.raises(litellm.BadRequestError) as excinfo:
await litellm.aimage_edit(
model=f"bedrock/{TEST_MODEL}",
prompt="restyle",
image=io.BytesIO(b"img-bytes"),
taskType="TEXT_IMAGE",
mask=io.BytesIO(b"mask-bytes"),
)
assert excinfo.value.status_code == 400
assert "does not support a mask" in str(excinfo.value)
async def test_aimage_edit_unsupported_task_type_maps_to_bad_request(monkeypatch):
"""Through the litellm image-edit layer, an unsupported taskType must surface as
litellm.BadRequestError (400-class), never APIConnectionError/500."""
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "env-bearer-token-12345")
with pytest.raises(litellm.BadRequestError) as excinfo:
await litellm.aimage_edit(
model=f"bedrock/{TEST_MODEL}",
prompt="restyle",
image=io.BytesIO(b"img-bytes"),
taskType="NOT_A_REAL_TASK_TYPE",
)
assert excinfo.value.status_code == 400
assert "Unsupported Amazon Nova Canvas taskType" in str(excinfo.value)
async def test_aimage_edit_forwards_style_to_nova_canvas_transform(monkeypatch):
"""style passed to litellm.aimage_edit must survive the images/main.py param
filtering and reach the Nova Canvas transform (regression: style used to be
blocklisted and silently dropped on the bedrock path)."""
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "env-bearer-token-12345")
posted: dict[str, object] = {}
class _FakeAsyncClient:
async def post(self, url, headers, data):
posted["url"] = url
posted["body"] = json.loads(data)
return httpx.Response(200, json={"images": ["aGk="]}, request=httpx.Request("POST", url))
import litellm.llms.bedrock.image_edit.handler as bedrock_image_edit_handler
with patch.object(
bedrock_image_edit_handler,
"get_async_httpx_client",
lambda **kwargs: _FakeAsyncClient(),
):
response = await litellm.aimage_edit(
model=f"bedrock/{TEST_MODEL}",
prompt="same layout",
image=io.BytesIO(b"img-bytes"),
taskType="TEXT_IMAGE",
style="DESIGN_SKETCH",
)
body = posted["body"]
assert isinstance(body, dict)
assert body["textToImageParams"]["style"] == "DESIGN_SKETCH"
assert response.data[0].b64_json == "aGk="
async def test_aimage_edit_image_less_condition_image_reaches_transform(monkeypatch):
"""POST /v1/images/edits with taskType TEXT_IMAGE and conditionImage only (no
multipart image) must reach the conditioned edit path instead of failing before
dispatch (aimage_edit used to require image as a positional argument)."""
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "env-bearer-token-12345")
posted: dict[str, object] = {}
class _FakeAsyncClient:
async def post(self, url, headers, data):
posted["url"] = url
posted["body"] = json.loads(data)
return httpx.Response(200, json={"images": ["aGk="]}, request=httpx.Request("POST", url))
import litellm.llms.bedrock.image_edit.handler as bedrock_image_edit_handler
with patch.object(
bedrock_image_edit_handler,
"get_async_httpx_client",
lambda **kwargs: _FakeAsyncClient(),
):
response = await litellm.aimage_edit(
model=f"bedrock/{TEST_MODEL}",
prompt="same layout",
taskType="TEXT_IMAGE",
conditionImage="aGVsbG8=",
)
body = posted["body"]
assert isinstance(body, dict)
assert body["taskType"] == "TEXT_IMAGE"
assert body["textToImageParams"]["conditionImage"] == "aGVsbG8="
assert response.data is not None
assert response.data[0].b64_json == "aGk="
def test_aimage_edit_positional_arguments_still_work(monkeypatch):
"""The pre-existing positional call form litellm.aimage_edit(image, model, prompt)
must keep working now that image is optional: the sync image_edit signature keeps
model/prompt defaulted and positional (no keyword-only marker), and aimage_edit
mirrors that compatibility profile while keeping its own parameter order."""
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "env-bearer-token-12345")
posted: dict[str, object] = {}
class _FakeAsyncClient:
async def post(self, url, headers, data):
posted["url"] = url
posted["body"] = json.loads(data)
return httpx.Response(200, json={"images": ["aGk="]}, request=httpx.Request("POST", url))
import litellm.llms.bedrock.image_edit.handler as bedrock_image_edit_handler
with patch.object(
bedrock_image_edit_handler,
"get_async_httpx_client",
lambda **kwargs: _FakeAsyncClient(),
):
response = asyncio.run(
litellm.aimage_edit(
io.BytesIO(b"img-bytes"),
f"bedrock/{TEST_MODEL}",
"same layout",
taskType="TEXT_IMAGE",
)
)
# Positional 2nd/3rd args landed on model/prompt respectively (the
# historical order), not swapped as sync's (image, prompt, model) order.
body = posted["body"]
assert isinstance(body, dict)
assert body["taskType"] == "TEXT_IMAGE"
assert body["textToImageParams"]["text"] == "same layout"
assert body["textToImageParams"]["conditionImage"] == base64.b64encode(b"img-bytes").decode("utf-8")
assert response.data is not None
assert response.data[0].b64_json == "aGk="
def test_aimage_edit_none_image_builds_empty_list(monkeypatch):
"""An omitted image must arrive at the handler as [] (empty list), not [None]."""
from unittest.mock import Mock
import litellm.images.main as images_main
seen: dict[str, object] = {}
def fake_image_edit(**kwargs):
seen.update(kwargs)
return Mock()
monkeypatch.setattr(images_main, "image_edit", fake_image_edit)
asyncio.run(
images_main.aimage_edit(
model=f"bedrock/{TEST_MODEL}",
prompt="same layout",
taskType="TEXT_IMAGE",
conditionImage="aGVsbG8=",
)
)
assert seen["image"] == []
#################################################
# logging headers redaction (pre_call additional_args)
#################################################
def test_redact_bedrock_headers_for_logging_masks_signed_headers():
"""SigV4 signature material must be replaced with [REDACTED]; safe headers survive."""
from litellm.llms.bedrock.common_utils import redact_bedrock_headers_for_logging
signed: Final[dict[str, str]] = {
"Content-Type": "application/json",
"Host": "bedrock-runtime.us-east-1.amazonaws.com",
"Authorization": (
"AWS4-HMAC-SHA256 Credential=AKIA-test/20260115/us-east-1/bedrock/aws4_request, "
"SignedHeaders=host;x-amz-date, Signature=deadbeefsecret"
),
"X-Amz-Date": "20260115T103000Z",
"X-Amz-Security-Token": "session-token-secret",
"X-Amzn-RequestId": "request-id-not-secret",
}
redacted = redact_bedrock_headers_for_logging(signed)
assert redacted["Content-Type"] == "application/json"
assert redacted["Host"] == "bedrock-runtime.us-east-1.amazonaws.com"
assert redacted["X-Amzn-RequestId"] == "request-id-not-secret"
assert redacted["Authorization"] == "[REDACTED]"
assert redacted["X-Amz-Date"] == "[REDACTED]"
assert redacted["X-Amz-Security-Token"] == "[REDACTED]"
# Every key stays present (log consumers see the full header shape).
assert set(redacted.keys()) == set(signed.keys())
# The input mapping is untouched: redaction never mutates the sent headers.
assert signed["Authorization"].startswith("AWS4-HMAC-SHA256")
assert signed["X-Amz-Security-Token"] == "session-token-secret"
def test_prepare_request_logging_headers_redacted(monkeypatch):
"""pre_call additional_args must carry the redacted copy; the sent request keeps
the real bearer Authorization header."""
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.llms.bedrock.image_edit.handler import BedrockImageEdit
monkeypatch.setenv("AWS_BEARER_TOKEN_BEDROCK", "env-bearer-token-12345")
captured: dict = {}
def _capture(model_call_details):
captured.update(model_call_details)
logging_obj = Logging(
model=f"bedrock/{TEST_MODEL}",
messages=[],
stream=False,
call_type="aimage_edit",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function",
kwargs={"logger_fn": _capture},
)
logging_obj.update_environment_variables(
litellm_params={"logger_fn": _capture},
optional_params={},
)
request = BedrockImageEdit()._prepare_request(
model=TEST_MODEL,
image=[io.BytesIO(b"fake-png")],
prompt="make it warmer",
optional_params={"aws_region_name": "us-west-2", "aws_profile_name": "litellm-no-such-aws-profile"},
api_base=None,
extra_headers=None,
logging_obj=logging_obj,
api_key=None,
)
logged_headers: Final = captured["additional_args"]["headers"]
assert logged_headers["Authorization"] == "[REDACTED]"
assert logged_headers["Content-Type"] == "application/json"
assert "env-bearer-token-12345" not in str(captured)
# The sent request still carries the real bearer Authorization header.
assert request.prepped.headers["Authorization"] == "Bearer env-bearer-token-12345"

View file

@ -2,13 +2,14 @@
import base64
import io
from typing import cast
from typing import Final, cast
from unittest.mock import Mock
import httpx
import pytest
import litellm
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock.image_edit.amazon_nova_canvas_image_edit_transformation import (
BedrockAmazonNovaCanvasImageEditConfig,
get_bedrock_image_edit_config_for_model,
@ -203,9 +204,7 @@ def test_transform_request_image_pathlike_input(tmp_path):
)
assert body["taskType"] == "IMAGE_VARIATION"
assert body["imageVariationParams"]["images"][0] == base64.b64encode(
image_bytes
).decode("utf-8")
assert body["imageVariationParams"]["images"][0] == base64.b64encode(image_bytes).decode("utf-8")
def test_transform_request_inpainting_with_mask():
@ -363,12 +362,10 @@ def test_transform_request_outpainting_without_mask_raises():
def test_transform_request_inpainting_explicit_task_without_mask_raises():
"""INPAINTING taskType without mask or maskPrompt must fail fast."""
"""INPAINTING taskType without mask or maskPrompt must fail fast with a 400-class error."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"img")
with pytest.raises(
ValueError, match="INPAINTING requires either maskPrompt or maskImage"
):
with pytest.raises(BedrockError, match="INPAINTING requires either maskPrompt or maskImage") as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="fix it",
@ -377,21 +374,23 @@ def test_transform_request_inpainting_explicit_task_without_mask_raises():
litellm_params={}, # type: ignore[arg-type]
headers={},
)
assert excinfo.value.status_code == 400
def test_transform_request_unknown_task_type_raises():
"""Unknown taskType must not silently map to IMAGE_VARIATION or INPAINTING."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"img")
with pytest.raises(ValueError, match="Unsupported Amazon Nova Canvas taskType"):
with pytest.raises(BedrockError, match="Unsupported Amazon Nova Canvas taskType") as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="x",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
image_edit_optional_request_params={"taskType": "NOT_A_REAL_TASK_TYPE"},
litellm_params={}, # type: ignore[arg-type]
headers={},
)
assert excinfo.value.status_code == 400
def test_transform_request_background_removal():
@ -410,6 +409,340 @@ def test_transform_request_background_removal():
assert "image" in body["backgroundRemovalParams"]
def test_transform_request_text_image_segmentation():
"""taskType TEXT_IMAGE + controlMode SEGMENTATION conditions via conditionImage (issue #39552)."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond-bytes")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="a city street in the same layout",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "SEGMENTATION",
"controlStrength": 0.7,
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
t2i = body["textToImageParams"]
assert t2i["text"] == "a city street in the same layout"
assert t2i["conditionImage"] == base64.b64encode(b"cond-bytes").decode("utf-8")
assert t2i["controlMode"] == "SEGMENTATION"
assert t2i["controlStrength"] == 0.7
def test_transform_request_text_image_canny_edge():
"""taskType TEXT_IMAGE + controlMode CANNY_EDGE keeps canny conditioning."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond-bytes")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="follow the edges",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "CANNY_EDGE",
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
t2i = body["textToImageParams"]
assert t2i["controlMode"] == "CANNY_EDGE"
assert t2i["conditionImage"]
assert "controlStrength" not in t2i
def test_transform_request_text_image_defaults_omit_control_fields():
"""Without controlMode/controlStrength only conditionImage is set (AWS defaults apply)."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="same composition",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
t2i = body["textToImageParams"]
assert t2i["conditionImage"]
assert "controlMode" not in t2i
assert "controlStrength" not in t2i
assert "negativeText" not in t2i
def test_transform_request_text_image_forwards_negative_text_and_igc():
"""TEXT_IMAGE forwards negativeText and allows imageGenerationConfig."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "SEGMENTATION",
"negativeText": "blurry",
"size": "1024x1024",
"seed": 7,
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["negativeText"] == "blurry"
assert body["imageGenerationConfig"]["width"] == 1024
assert body["imageGenerationConfig"]["height"] == 1024
assert body["imageGenerationConfig"]["seed"] == 7
def test_transform_request_text_image_invalid_control_mode_raises():
"""Unknown controlMode fails fast instead of hitting AWS with a bad payload."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "CANNY_EDIT",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "Unsupported Amazon Nova Canvas controlMode" in str(excinfo.value.message)
def test_transform_request_text_image_with_mask_raises():
"""TEXT_IMAGE has no mask field; a provided mask must fail fast."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
mask = io.BytesIO(b"mask-bytes")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"mask": mask,
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "does not support a mask" in str(excinfo.value.message)
assert "INPAINTING or OUTPAINTING" in str(excinfo.value.message)
def test_transform_request_text_image_with_mask_prompt_raises():
"""TEXT_IMAGE has no maskPrompt field either; fail fast like the binary mask
instead of silently dropping it."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"maskPrompt": "the sky region",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "does not support a mask" in str(excinfo.value.message)
assert "INPAINTING or OUTPAINTING" in str(excinfo.value.message)
def test_transform_request_image_variation_with_mask_still_ignores_mask():
"""The documented IMAGE_VARIATION mask-ignoring behavior is untouched by the guard."""
config = BedrockAmazonNovaCanvasImageEditConfig()
main = io.BytesIO(b"img-bytes")
mask = io.BytesIO(b"mask-bytes")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="vary style",
image=main,
image_edit_optional_request_params={
"taskType": "IMAGE_VARIATION",
"mask": mask,
},
litellm_params={},
headers={},
)
assert body["taskType"] == "IMAGE_VARIATION"
assert "maskImage" not in body["imageVariationParams"]
assert body["imageVariationParams"]["text"] == "vary style"
def test_transform_request_text_image_empty_prompt_raises():
"""An empty TEXT_IMAGE prompt must fail fast, not silently send a blank."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "TEXT_IMAGE requires a text prompt" in str(excinfo.value.message)
def test_transform_request_text_image_none_prompt_raises():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt=None,
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "TEXT_IMAGE requires a text prompt" in str(excinfo.value.message)
def test_transform_request_text_image_control_strength_out_of_range_raises():
"""controlStrength outside 0.0-1.0 fails fast."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": 1.5,
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "controlStrength must be between 0.0 and 1.0" in str(excinfo.value.message)
@pytest.mark.parametrize("control_strength", [0.0, 1.0])
def test_transform_request_text_image_control_strength_bounds_pass(control_strength):
"""Boundary controlStrength values 0.0 and 1.0 are accepted."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": control_strength,
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["controlStrength"] == control_strength
def test_transform_request_text_image_control_strength_string_coerced():
"""controlStrength arriving as a string (multipart form data) is coerced to a float."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": "0.7",
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["controlStrength"] == 0.7
def test_transform_request_text_image_control_strength_non_numeric_string_raises():
"""A non-numeric controlStrength string must fail fast, not TypeError."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlStrength": "abc",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "controlStrength must be a number" in str(excinfo.value.message)
def test_transform_request_text_image_forwards_style():
"""style is forwarded into textToImageParams for conditioned editing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="a city street in the same layout",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "SEGMENTATION",
"style": "DESIGN_SKETCH",
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
assert body["textToImageParams"]["style"] == "DESIGN_SKETCH"
def test_transform_request_text_image_omits_style_when_absent():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={"taskType": "TEXT_IMAGE"},
litellm_params={},
headers={},
)
assert "style" not in body["textToImageParams"]
def test_get_supported_openai_params_includes_conditioning_fields():
"""controlMode/controlStrength are advertised for TEXT_IMAGE conditioned editing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
supported = config.get_supported_openai_params("amazon.nova-canvas-v1:0")
assert "controlMode" in supported
assert "controlStrength" in supported
def test_get_supported_openai_params_includes_style():
"""style is advertised for TEXT_IMAGE conditioned editing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
supported = config.get_supported_openai_params("amazon.nova-canvas-v1:0")
assert "style" in supported
def test_transform_request_background_removal_omits_image_generation_config():
"""AWS Nova Canvas does not allow imageGenerationConfig on BACKGROUND_REMOVAL."""
config = BedrockAmazonNovaCanvasImageEditConfig()
@ -627,3 +960,161 @@ def test_prepare_request_bearer_token_never_runs_the_sigv4_credential_chain(monk
)
assert request.prepped.headers["Authorization"] == "Bearer env-bearer-token-12345"
def test_transform_request_conditioning_fields_without_text_image_task_type_raises():
"""controlMode/controlStrength/style with any non-TEXT_IMAGE resolved taskType must
fail fast instead of being silently dropped by mask/variation routing."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"controlMode": "SEGMENTATION",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "taskType=TEXT_IMAGE" in str(excinfo.value.message)
assert "controlMode/controlStrength/style" in str(excinfo.value.message)
def test_transform_request_control_strength_without_text_image_task_type_raises():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"controlStrength": 0.4,
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "taskType=TEXT_IMAGE" in str(excinfo.value.message)
def test_transform_request_style_without_text_image_task_type_raises():
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"style": "DESIGN_SKETCH",
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "taskType=TEXT_IMAGE" in str(excinfo.value.message)
def test_transform_request_conditioning_fields_with_explicit_text_image_works():
"""The guard must not fire when taskType is explicitly TEXT_IMAGE."""
config = BedrockAmazonNovaCanvasImageEditConfig()
img = io.BytesIO(b"cond")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=img,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"controlMode": "CANNY_EDGE",
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
assert body["textToImageParams"]["controlMode"] == "CANNY_EDGE"
def test_transform_request_condition_image_without_multipart_image():
"""conditionImage is an alternative TEXT_IMAGE condition source for JSON-only callers."""
config = BedrockAmazonNovaCanvasImageEditConfig()
condition_b64: Final = base64.b64encode(b"cond-bytes").decode("utf-8")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="same layout",
image=None,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"conditionImage": condition_b64,
"controlMode": "SEGMENTATION",
},
litellm_params={},
headers={},
)
assert body["taskType"] == "TEXT_IMAGE"
t2i = body["textToImageParams"]
assert t2i["conditionImage"] == condition_b64
assert t2i["controlMode"] == "SEGMENTATION"
def test_transform_request_condition_image_bytes_accepted():
"""Raw bytes conditionImage is base64-encoded on the way into the body."""
config = BedrockAmazonNovaCanvasImageEditConfig()
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="same layout",
image=None,
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"conditionImage": b"raw-cond-bytes",
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["conditionImage"] == base64.b64encode(b"raw-cond-bytes").decode("utf-8")
def test_transform_request_multipart_image_wins_over_condition_image():
"""Pinned precedence: when both are supplied the multipart `image` field wins."""
config = BedrockAmazonNovaCanvasImageEditConfig()
condition_b64: Final = base64.b64encode(b"from-condition-param").decode("utf-8")
body, _ = config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="same layout",
image=io.BytesIO(b"from-multipart"),
image_edit_optional_request_params={
"taskType": "TEXT_IMAGE",
"conditionImage": condition_b64,
},
litellm_params={},
headers={},
)
assert body["textToImageParams"]["conditionImage"] == base64.b64encode(b"from-multipart").decode("utf-8")
def test_transform_request_condition_image_with_other_task_type_raises():
"""conditionImage only conditions TEXT_IMAGE; other task types need the image input."""
config = BedrockAmazonNovaCanvasImageEditConfig()
with pytest.raises(BedrockError) as excinfo:
config.transform_image_edit_request(
model="amazon.nova-canvas-v1:0",
prompt="restyle",
image=None,
image_edit_optional_request_params={
"taskType": "IMAGE_VARIATION",
"conditionImage": base64.b64encode(b"cond").decode("utf-8"),
},
litellm_params={},
headers={},
)
assert excinfo.value.status_code == 400
assert "conditionImage is only supported" in str(excinfo.value.message)
def test_get_supported_openai_params_includes_condition_image():
config = BedrockAmazonNovaCanvasImageEditConfig()
supported = config.get_supported_openai_params("amazon.nova-canvas-v1:0")
assert "conditionImage" in supported

File diff suppressed because it is too large Load diff