import base64 import json import os import time from collections.abc import Iterable, Mapping, MutableMapping, Sequence from contextlib import suppress from functools import cache from itertools import chain from types import MappingProxyType from typing import Any, Final, TypeAlias, TypedDict from urllib.parse import unquote import httpx from httpx import Headers, Response from openai.types.file_deleted import FileDeleted from pydantic import BaseModel, ConfigDict, TypeAdapter from typing_extensions import ReadOnly from litellm._logging import verbose_logger from litellm._uuid import uuid from litellm.constants import BEDROCK_INVOKE_PROVIDERS_LITERAL from litellm.files.utils import FilesAPIUtils from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix from litellm.litellm_core_utils.cloud_storage_security import ( BEDROCK_MANAGED_S3_BATCH_PREFIX, BEDROCK_MANAGED_S3_PREFIXES, BEDROCK_MANAGED_S3_UPLOAD_PREFIX, build_managed_cloud_object_name, encode_s3_object_key_for_url, sanitize_cloud_object_component, should_allow_legacy_cloud_file_ids, split_configured_cloud_bucket_name, validate_managed_cloud_file_id, ) from litellm.litellm_core_utils.prompt_templates.common_utils import ( extract_file_data, text_completion_prompt_to_messages, ) from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.llms.base_llm.files.transformation import ( BaseFilesConfig, LiteLLMLoggingObj, ) from litellm.types.llms.bedrock import BedrockBatchRecordKind from litellm.types.llms.openai import ( AllMessageValues, CreateFileRequest, FileContentRequest, FileTypes, HttpxBinaryResponseContent, OpenAICreateFileRequestOptionalParams, OpenAIFileObject, PathLike, ResponseInputParam, ResponsesAPIOptionalRequestParams, ) from litellm.types.utils import ExtractedFileData, LlmProviders, SpecialEnums from litellm.utils import get_llm_provider from ..base_aws_llm import BaseAWSLLM from ..common_utils import BedrockError, merge_bedrock_aws_request_params, resolve_s3_encryption_key_id # litellm_params key used to hand the SigV4-signed GET headers from # `transform_file_content_request` to `validate_environment` (the only hook # the shared file-content HTTP handler exposes for setting request headers). # Same pattern as the `upload_url` handoff in `transform_create_file_request`. S3_SIGNED_GET_HEADERS_PARAM: Final = "_s3_signed_get_headers" # litellm_params key carrying the size of the body uploaded to S3, handed from # `transform_create_file_request` to `transform_create_file_response`. UPLOAD_CONTENT_LENGTH_PARAM: Final = "_s3_upload_content_length" def _frozen_mapping(items: Iterable[tuple[str, object]]) -> Mapping[str, object]: return MappingProxyType(dict(items)) def _strip_llm_routing_prefix(model: str) -> str: try: stripped_model, _, _, _ = get_llm_provider(model=model, custom_llm_provider=None) except Exception as e: verbose_logger.exception( "litellm.llms.bedrock.files.transformation.py::_strip_llm_routing_prefix() - Error inferring custom_llm_provider - %s", e, ) return model return stripped_model _EmbeddingBatchInput: TypeAlias = ( str | int | float | Sequence[str] | Sequence[int] | Sequence[Sequence[int]] | Mapping[str, object] ) class _OpenAIBatchRecordBody(TypedDict, total=False): model: ReadOnly[str] prompt: ReadOnly[str | Sequence[str] | Sequence[int] | Sequence[Sequence[int]]] input: ReadOnly[_EmbeddingBatchInput] metadata: ReadOnly[Mapping[str, object]] class _OpenAIBatchRecord(TypedDict, total=False): custom_id: ReadOnly[str] url: ReadOnly[str] body: ReadOnly[_OpenAIBatchRecordBody] class _BedrockBatchRecord(TypedDict): recordId: ReadOnly[str] modelInput: ReadOnly[Mapping[str, object]] class _S3UploadResponse(TypedDict, total=False): Key: ReadOnly[str] Bucket: ReadOnly[str] ContentLength: ReadOnly[int] # JSONL batch records are untyped json, so the `/v1/responses` fields are # validated into their concrete Responses API types before being handed to the # Responses-to-Chat bridge. Both adapters drop keys the Responses API doesn't # define, which is what the bridge would ignore anyway. Built on first use # rather than at import: `ResponseInputParam` is a deep union and only batch # files carrying `/v1/responses` records need it. @cache def _responses_input_adapter() -> TypeAdapter[str | ResponseInputParam]: return TypeAdapter(str | ResponseInputParam) @cache def _responses_request_adapter() -> TypeAdapter[ResponsesAPIOptionalRequestParams]: return TypeAdapter(ResponsesAPIOptionalRequestParams) class _BedrockS3RequestParams(BaseModel): """Typed view of the credential/region params the S3 GetObject path reads.""" model_config = ConfigDict(extra="ignore") aws_access_key_id: str | None = None aws_secret_access_key: str | None = None aws_session_token: str | None = None aws_region_name: str | None = None aws_session_name: str | None = None aws_profile_name: str | None = None aws_role_name: str | None = None aws_web_identity_token: str | None = None aws_sts_endpoint: str | None = None s3_region_name: str | None = None s3_endpoint_url: str | None = None class _TrustedS3ModelCredentials(BaseModel): """The S3 buckets the server trusts file ids against, from the deployment snapshot.""" model_config = ConfigDict(extra="ignore") s3_bucket_name: str | None = None s3_output_bucket_name: str | None = None def extract_s3_uri_from_file_id(file_id: str) -> str: """ Resolve a Bedrock file id to its S3 URI. Accepts either a base64-encoded LiteLLM unified file id (whose decoded form carries `llm_output_file_id,s3://...`) or a direct `s3://` URI. """ try: padded: Final = file_id + "=" * (-len(file_id) % 4) decoded: Final = base64.urlsafe_b64decode(padded).decode() if decoded.startswith(SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value): if "llm_output_file_id," in decoded: return decoded.split("llm_output_file_id,")[1].split(";")[0] except Exception: pass if file_id.startswith("s3://"): return file_id raise ValueError("file_id must be a managed LiteLLM S3 file id") _S3_BUCKET_REQUIRED_ERROR: Final = "S3 bucket_name is required. Set 's3_bucket_name' in proxy config or AWS_S3_BUCKET_NAME for Bedrock file content retrieval." def _trusted_s3_model_credentials(litellm_params: Mapping[str, object]) -> _TrustedS3ModelCredentials: trusted_model_credentials: Final = litellm_params.get("_litellm_internal_model_credentials") if not isinstance(trusted_model_credentials, MappingProxyType): return _TrustedS3ModelCredentials() snapshot: Final[dict[str, object]] = {} snapshot.update(trusted_model_credentials) # any-ok: untyped snapshot return _TrustedS3ModelCredentials.model_validate(snapshot) def get_configured_s3_bucket_name(litellm_params: Mapping[str, object]) -> str: """ Resolve the server-configured S3 bucket for Bedrock file operations. Only trusts the immutable server-side credential snapshot or the environment; never a request-supplied param, since the bucket is what `validate_managed_cloud_file_id` checks file ids against. """ bucket_name: Final = _trusted_s3_model_credentials(litellm_params).s3_bucket_name or os.getenv("AWS_S3_BUCKET_NAME") if not bucket_name: raise ValueError(_S3_BUCKET_REQUIRED_ERROR) return bucket_name def get_configured_s3_bucket_names(litellm_params: Mapping[str, object]) -> tuple[str, ...]: """ Resolve the server-configured S3 buckets a Bedrock file id may live in. Bedrock batch outputs land in ``s3_output_bucket_name`` when it differs from the input bucket, so retrieval validates against both. Same trust rules as ``get_configured_s3_bucket_name``: only the immutable credential snapshot or the environment, never a request param. """ trusted: Final = _trusted_s3_model_credentials(litellm_params) input_bucket: Final = trusted.s3_bucket_name or os.getenv("AWS_S3_BUCKET_NAME") output_bucket: Final = trusted.s3_output_bucket_name or os.getenv("AWS_S3_OUTPUT_BUCKET_NAME") buckets: Final = tuple(dict.fromkeys(bucket for bucket in (input_bucket, output_bucket) if bucket)) if not buckets: raise ValueError(_S3_BUCKET_REQUIRED_ERROR) return buckets def _validate_file_id_against_configured_buckets( s3_uri: str, configured_bucket_names: tuple[str, ...], allow_legacy_cloud_file_ids: bool, ) -> tuple[str, str]: def validate_against(configured_bucket_name: str) -> tuple[str, str]: return validate_managed_cloud_file_id( file_id=s3_uri, scheme="s3://", configured_bucket_name=configured_bucket_name, allowed_object_prefixes=BEDROCK_MANAGED_S3_PREFIXES, allow_legacy_cloud_file_ids=allow_legacy_cloud_file_ids, ) for candidate_bucket_name in configured_bucket_names[:-1]: with suppress(ValueError): return validate_against(candidate_bucket_name) return validate_against(configured_bucket_names[-1]) def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Response) -> int: """ S3 answers PutObject with an empty body, so the stored object size comes from the signed request recorded by `transform_create_file_request`, not the response headers. """ uploaded_size: Final = litellm_params.get(UPLOAD_CONTENT_LENGTH_PARAM) if isinstance(uploaded_size, int): return uploaded_size response_content_length: Final = raw_response.headers.get("Content-Length", "0") return int(response_content_length) if response_content_length.isdigit() else 0 class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig): """ Config for Bedrock Files - handles S3 uploads for Bedrock batch processing """ def __init__(self): self.jsonl_transformation = BedrockJsonlFilesTransformation() super().__init__() @property def custom_llm_provider(self) -> LlmProviders: return LlmProviders.BEDROCK @property def file_upload_http_method(self) -> str: """ Bedrock files are uploaded to S3, which requires PUT requests """ return "PUT" def validate_environment( self, headers: MutableMapping[str, object], model: str, messages: list[AllMessageValues], optional_params: dict, litellm_params: MutableMapping[str, object], api_key: str | None = None, api_base: str | None = None, ) -> dict: result: Final[dict[str, object]] = {} result.update(headers) signed_headers: Final = litellm_params.pop(S3_SIGNED_GET_HEADERS_PARAM, None) if isinstance(signed_headers, Mapping): result.update(signed_headers) # any-ok: untyped handoff headers # otherwise no extra headers - AWS credentials are handled by BaseAWSLLM return result def _get_content_from_openai_file(self, openai_file_content: FileTypes) -> str: """ Helper to extract content from various OpenAI file types and return as string. Handles: - Direct content (str, bytes, IO[bytes]) - Tuple formats: (filename, content, [content_type], [headers]) - PathLike objects """ content: str | bytes = b"" # Extract file content from tuple if necessary if isinstance(openai_file_content, tuple): # Take the second element which is always the file content file_content = openai_file_content[1] else: file_content = openai_file_content # Handle different file content types if isinstance(file_content, str): # String content can be used directly content = file_content elif isinstance(file_content, bytes): # Bytes content can be decoded content = file_content elif isinstance(file_content, PathLike): # PathLike with open(str(file_content), "rb") as f: content = f.read() elif hasattr(file_content, "read"): # IO[bytes] # File-like objects need to be read content = file_content.read() # Ensure content is string if isinstance(content, bytes): content = content.decode("utf-8") return content def _get_s3_object_name_from_batch_jsonl( self, openai_jsonl_content: Sequence[_OpenAIBatchRecord], ) -> str: """ Gets a unique S3 object name for the Bedrock batch processing job named as: litellm-bedrock-files/{model}/{uuid} """ _model = openai_jsonl_content[0].get("body", {}).get("model", "") # Remove bedrock/ prefix if present _model = _model.removeprefix("bedrock/") safe_model: Final = sanitize_cloud_object_component(_model.replace(":", "-"), fallback="model") object_name: Final = f"{BEDROCK_MANAGED_S3_BATCH_PREFIX}{safe_model}-{uuid.uuid4()}.jsonl" return object_name def get_object_name(self, extracted_file_data: ExtractedFileData, purpose: str) -> str: """ Get the object name for the request """ extracted_file_data_content: Final = extracted_file_data.get("content") if extracted_file_data_content is None: raise ValueError("file content is required") if purpose == "batch": ## 1. If jsonl, check if there's a model name file_content: Final = self._get_content_from_openai_file(extracted_file_data_content) # Split into lines and parse each line as JSON openai_jsonl_content: Final = [json.loads(line) for line in file_content.splitlines() if line.strip()] if len(openai_jsonl_content) > 0: return self._get_s3_object_name_from_batch_jsonl(openai_jsonl_content) ## 2. If not jsonl, store under a server-generated managed object name filename: Final = extracted_file_data.get("filename") return build_managed_cloud_object_name( prefix=BEDROCK_MANAGED_S3_UPLOAD_PREFIX, filename=filename, fallback_filename="file", ) def get_complete_file_url( self, api_base: str | None, api_key: str | None, model: str, optional_params: dict, litellm_params: dict, data: CreateFileRequest, ) -> str: """ Get the complete S3 URL for the file upload request """ request_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params) bucket_name = litellm_params.get("s3_bucket_name") or os.getenv("AWS_S3_BUCKET_NAME") if not bucket_name: raise ValueError( "S3 bucket_name is required. Set 's3_bucket_name' in litellm_params or AWS_S3_BUCKET_NAME env var" ) bucket_name, object_prefix = split_configured_cloud_bucket_name(bucket_name) s3_region_name: Final = litellm_params.get("s3_region_name") or optional_params.get("s3_region_name") aws_region_name: Final = s3_region_name or self._get_aws_region_name(request_params, model) file_data: Final = data.get("file") purpose: Final = data.get("purpose") if file_data is None: raise ValueError("file is required") if purpose is None: raise ValueError("purpose is required") extracted_file_data: Final = extract_file_data(file_data) object_name = self.get_object_name(extracted_file_data, purpose) if object_prefix: object_name = f"{object_prefix}/{object_name}" encoded_object_name: Final = encode_s3_object_key_for_url(object_name) # S3 endpoint URL format s3_endpoint_url: Final = ( request_params.get("s3_endpoint_url") or f"https://s3.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}" ).rstrip("/") return f"{s3_endpoint_url}/{bucket_name}/{encoded_object_name}" def get_supported_openai_params(self, model: str) -> list[OpenAICreateFileRequestOptionalParams]: return [] def map_openai_params( self, non_default_params: dict, optional_params: dict, model: str, drop_params: bool, ) -> dict: return optional_params # Providers whose InvokeModel body uses the Converse API format # (messages + inferenceConfig + image blocks). Nova is the primary # example; add others here as they adopt the same schema. CONVERSE_INVOKE_PROVIDERS = ("nova",) # OpenAI batch URLs that select which request shape a JSONL line carries. # Per the OpenAI Batch API spec every record carries a `url`, so we use it # as the authoritative routing signal instead of inferring from the # presence of `input` vs `prompt` vs `messages`. OPENAI_EMBEDDINGS_URL = "/v1/embeddings" OPENAI_TEXT_COMPLETIONS_URL = "/v1/completions" OPENAI_RESPONSES_URL = "/v1/responses" @staticmethod def _classify_batch_record(openai_jsonl_record: _OpenAIBatchRecord) -> BedrockBatchRecordKind: """ Decide which OpenAI endpoint shape an OpenAI batch JSONL line carries. Precedence (strict - any recognized `url` short-circuits): 1. A `url` matching a supported endpoint wins. Authoritative per the OpenAI Batch API spec, which requires it on every record. 2. Any other non-empty `url` -> chat. We trust the caller's explicit signal rather than re-deriving it from the body, and an unexpectedly-shaped body fails cleanly inside the chat transformer instead of being silently misrouted. 3. `url` missing/empty -> fall back to body shape. `messages` wins over the other keys so a malformed record carrying several of them keeps its conversation instead of having it dropped, and a bare `input` stays an embedding for backwards compatibility (that ambiguity with `/v1/responses` is only resolvable from `url`). """ match openai_jsonl_record.get("url"): case BedrockFilesConfig.OPENAI_EMBEDDINGS_URL: return BedrockBatchRecordKind.EMBEDDING case BedrockFilesConfig.OPENAI_TEXT_COMPLETIONS_URL: return BedrockBatchRecordKind.TEXT_COMPLETION case BedrockFilesConfig.OPENAI_RESPONSES_URL: return BedrockBatchRecordKind.RESPONSES case None | "": pass case _: return BedrockBatchRecordKind.CHAT body: Final = openai_jsonl_record.get("body") if not isinstance(body, Mapping): return BedrockBatchRecordKind.CHAT if "messages" in body: return BedrockBatchRecordKind.CHAT if "prompt" in body: return BedrockBatchRecordKind.TEXT_COMPLETION if "input" in body: return BedrockBatchRecordKind.EMBEDDING return BedrockBatchRecordKind.CHAT # Identifier for the Bedrock Titan v2 InvokeModel body schema as stored # in `model_prices_and_context_window.json`. Centralized so future # embedding-schema variants can add their own value # (e.g. `cohere_v3`, `titan_g1`, `titan_multimodal`) without touching # the detection logic. _TITAN_V2_INVOCATION_SCHEMA = "titan_v2" # Substring marker used as a fallback when the registry can't resolve # the model id - notably cross-region inference profile prefixes # (`us.amazon.titan-embed-text-v2:0`) and Bedrock ARN forms, which # `get_model_info` doesn't normalize today. _TITAN_V2_EMBED_MODEL_MARKER = "titan-embed-text-v2" # Nested field name under `provider_specific_entry` that identifies the # Bedrock InvokeModel body schema for batch inference. # `provider_specific_entry` is the registry's escape hatch for fields # `get_model_info` doesn't promote to top-level - exactly what we need # here. Documented in the `sample_spec` entry of # `model_prices_and_context_window.json` and surfaced by # `get_model_info` (see `ModelInfo.provider_specific_entry`). _BEDROCK_INVOCATION_SCHEMA_FIELD = "bedrock_invocation_schema" @staticmethod def _is_titan_v2_embed_model(model: str) -> bool: """ True iff `model` refers to Amazon Titan Text Embeddings V2. Resolution order: 1. `model_prices_and_context_window.json` via `get_model_info`. The Titan v2 registry entry carries an explicit `provider_specific_entry.bedrock_invocation_schema` discriminator (`"titan_v2"`). When the registry resolves the id we trust that field as the source of truth - no hardcoded model-id comparison needed. 2. Substring fallback (`titan-embed-text-v2` followed by `:`, `/`, or end-of-string) for ids the registry can't normalize. This catches cross-region inference profile prefixes (`us.amazon.titan-embed-text-v2:0`) and Bedrock ARN forms; the marker boundary check rejects lookalikes like `titan-embed-text-v20` or `titan-embed-text-v2-experimental`. Tolerant of common id shapes: - "amazon.titan-embed-text-v2:0" - "bedrock/amazon.titan-embed-text-v2:0" - "us.amazon.titan-embed-text-v2:0" (cross-region inference profile) - ARN forms ending in ".../amazon.titan-embed-text-v2:0" """ # Registry-driven path: when get_model_info resolves the id we trust # the registry's discriminator. A resolved id with a different (or # absent) schema value here is intentionally not given a substring # second-chance - the registry is authoritative for ids it knows. registry_schema: Final = BedrockFilesConfig._lookup_provider_specific_field( model, BedrockFilesConfig._BEDROCK_INVOCATION_SCHEMA_FIELD ) if registry_schema is not None: return registry_schema == BedrockFilesConfig._TITAN_V2_INVOCATION_SCHEMA # Registry silence -> substring fallback for unmapped ids only. normalized = model.lower() normalized = normalized.removeprefix("bedrock/") marker: Final = BedrockFilesConfig._TITAN_V2_EMBED_MODEL_MARKER idx: Final = normalized.find(marker) if idx < 0: return False end: Final = idx + len(marker) return end == len(normalized) or normalized[end] in (":", "/") @staticmethod def _lookup_provider_specific_field(model_id: str, field: str) -> str | None: """ Read a nested string field from the registry entry's `provider_specific_entry` dict via `litellm.get_model_info`. Returns the field's string value when: - the registry resolves `model_id`, - the entry exposes `provider_specific_entry` as a dict, and - that dict has `field` mapped to a non-empty string. Otherwise returns `None`. Isolating this means feature detectors (Titan v2 today, future Cohere Embed / Nova Multimodal branches) share one defensive try/except shape instead of duplicating it. The `None` return covers every realistic failure mode: `get_model_info` raises (cross-region profile prefixes, Bedrock ARN forms, unreleased models), returns a non-dict, has no `provider_specific_entry`, or the requested field is missing / non-string / empty. """ try: from litellm import get_model_info info: Final = get_model_info(model_id) except Exception: return None if not isinstance(info, dict): return None provider_specific: Final = info.get("provider_specific_entry") if not isinstance(provider_specific, dict): return None value: Final = provider_specific.get(field) return value if isinstance(value, str) and value else None @staticmethod def _coerce_embedding_input_to_string(raw_input: _EmbeddingBatchInput | None, model: str = "") -> str: """ Normalize an OpenAI /v1/embeddings `input` field into the single string that Bedrock Titan v2 InvokeModel expects in `inputText`. Accepts: a string, or a single-element list containing one string. Rejects (with actionable messages): - None / missing -> ValueError - Multi-element string lists -> ValueError, prompts caller to emit one JSONL line per input - Pre-tokenized inputs (List[int], List[List[int]]) -> NotImplementedError - Any other type -> ValueError Extracted so the validation can be exercised in isolation and so future embedding-provider branches (Titan G1, Cohere) can reuse it without duplicating the type-shaping logic. """ if raw_input is None: raise ValueError(f"Embedding batch record is missing required `input` field: model={model}") # Bedrock InvokeModel for Titan v2 takes exactly one string `inputText` # per call. Pre-tokenized inputs and multi-element string lists are # explicitly unsupported so callers emit one JSONL line per embedding # instead of relying on us to silently fan out or concatenate. if isinstance(raw_input, list): if len(raw_input) == 1: candidate = raw_input[0] else: raise ValueError( "Bedrock batch embedding requires one input per JSONL " "record. Got a list with " f"{len(raw_input)} items for model={model}; emit one " "JSONL line per input string instead." ) else: candidate = raw_input # Catches pre-tokenized inputs (List[int] from OpenAI spec, or a # single int slipping past the list-unwrap above). # NOTE: bool is a subclass of int but treating True/False as a token # is meaningless either way, so the broad check is fine. if isinstance(candidate, (list, int)): raise NotImplementedError( "Bedrock Titan v2 batch embedding does not support " "pre-tokenized integer inputs. Pass `input` as a string " f"(model={model})." ) if not isinstance(candidate, str): raise ValueError( "Bedrock batch embedding `input` must be a string (or a " "single-element list of strings). Got type " f"{type(candidate).__name__} for model={model}." ) return candidate def _map_openai_embedding_to_bedrock_params( self, openai_request_body: _OpenAIBatchRecordBody, model: str, ) -> dict[str, object]: """ Transform an OpenAI /v1/embeddings request body into the Bedrock InvokeModel `modelInput` for embedding models that AWS supports via batch inference (CreateModelInvocationJob). Currently routes Amazon Titan Text Embeddings V2 only; other embedding providers (Titan G1, Titan Multimodal, Cohere Embed, Nova Multimodal Embeddings) raise NotImplementedError until they get a dedicated branch. Splitting them keeps PR scope tight and lets each model's request schema be exercised by its own tests. AWS docs (Titan v2 InvokeModel body): https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-titan-embed-text.html """ from litellm.llms.bedrock.embed.amazon_titan_v2_transformation import ( AmazonTitanV2Config, ) if not self._is_titan_v2_embed_model(model): # Refuse early instead of silently shaping the body for the wrong # provider. The synchronous /v1/embeddings path supports more # models, but each has a different InvokeModel schema; mapping # them here without dedicated tests would risk corrupt batches. raise NotImplementedError( "Bedrock batch embedding currently supports only Amazon " "Titan Text Embeddings V2 (model id contains " f"'titan-embed-text-v2'). Got model={model!r}. Track other " "embedding models in https://github.com/BerriAI/litellm/issues." ) input_text: Final = self._coerce_embedding_input_to_string(openai_request_body.get("input"), model=model) # Map OpenAI-style params (dimensions, encoding_format) onto the # Titan v2 schema (dimensions, embeddingTypes) via the embed config # so this stays in sync with the synchronous /v1/embeddings path. non_default_params: Final = {k: v for k, v in openai_request_body.items() if k not in ("model", "input")} titan_config: Final = AmazonTitanV2Config() inference_params: Final = titan_config.map_openai_params( non_default_params=non_default_params, optional_params={}, ) return dict(titan_config._transform_request(input=input_text, inference_params=inference_params)) @staticmethod def _transform_text_completion_body_to_chat_body( openai_request_body: _OpenAIBatchRecordBody, ) -> Mapping[str, object]: """ Rewrite an OpenAI `/v1/completions` batch body as a Chat Completions body. Bedrock batch `modelInput` is the model's InvokeModel/Converse body, and no Bedrock batch model takes a bare `prompt`, so the wrapping that `litellm.text_completion` does in real time has to happen here too. """ prompt: Final = openai_request_body.get("prompt") if prompt is None: raise ValueError( "Batch record for /v1/completions is missing required `prompt` field: " f"model={openai_request_body.get('model', '')}" ) return _frozen_mapping( chain( ((key, value) for key, value in openai_request_body.items() if key != "prompt"), (("messages", text_completion_prompt_to_messages(prompt)),), ) ) @staticmethod def _transform_responses_body_to_chat_body(openai_request_body: _OpenAIBatchRecordBody) -> Mapping[str, object]: """ Rewrite an OpenAI `/v1/responses` batch body as a Chat Completions body. Delegates to the same Responses-to-Chat bridge the real-time path uses for providers without a native Responses API (which is every Bedrock model), so `input`, `instructions`, `max_output_tokens` and the tool params translate identically in batch and real time. The bridge always emits a `tools` key; an empty one is dropped rather than shipped as an empty array inside `modelInput`. """ from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) responses_input: Final = openai_request_body.get("input") if responses_input is None: raise ValueError( "Batch record for /v1/responses is missing required `input` field: " f"model={openai_request_body.get('model', '')}" ) chat_body: Final[Mapping[str, object]] = ( LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( model=openai_request_body.get("model", ""), input=_responses_input_adapter().validate_python(responses_input), responses_api_request=_responses_request_adapter().validate_python( _frozen_mapping( (key, value) for key, value in openai_request_body.items() if key not in ("model", "input") ) ), metadata=openai_request_body.get("metadata"), ) ) return _frozen_mapping((key, value) for key, value in chat_body.items() if key != "tools" or value) @staticmethod def _transform_batch_body_to_chat_body( openai_request_body: _OpenAIBatchRecordBody, record_kind: BedrockBatchRecordKind, ) -> Mapping[str, object]: """ Normalize a non-embedding batch body to the Chat Completions shape the per-provider Bedrock transformations expect. """ match record_kind: case BedrockBatchRecordKind.TEXT_COMPLETION: return BedrockFilesConfig._transform_text_completion_body_to_chat_body(openai_request_body) case BedrockBatchRecordKind.RESPONSES: return BedrockFilesConfig._transform_responses_body_to_chat_body(openai_request_body) case BedrockBatchRecordKind.CHAT: return openai_request_body case BedrockBatchRecordKind.EMBEDDING: raise ValueError("Embedding batch records do not have a chat-completion equivalent") def _map_openai_to_bedrock_params( self, openai_request_body: Mapping[str, Any], model: str, provider: str | None = None, ) -> dict[str, object]: """ Transform OpenAI request body to Bedrock-compatible modelInput parameters using existing transformation logic. Routes to the correct per-provider transformation so that the resulting dict matches the InvokeModel body that Bedrock expects for batch inference. """ from litellm.types.utils import LlmProviders messages: Final = openai_request_body.get("messages", []) optional_params: Final = {k: v for k, v in openai_request_body.items() if k not in ["model", "messages"]} # --- Anthropic: use existing AmazonAnthropicClaudeConfig --- if provider == LlmProviders.ANTHROPIC: from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import ( AmazonAnthropicClaudeConfig, ) config: Final = AmazonAnthropicClaudeConfig() mapped_params = config.map_openai_params( non_default_params={}, optional_params=optional_params, model=model, drop_params=False, ) return config.transform_request( model=model, messages=messages, optional_params=mapped_params, litellm_params={}, headers={}, ) # --- Converse API providers (e.g. Nova): use AmazonConverseConfig # to correctly convert image_url blocks to Bedrock image format # and wrap inference params inside inferenceConfig. --- if provider in self.CONVERSE_INVOKE_PROVIDERS: from litellm.llms.bedrock.chat.converse_transformation import ( AmazonConverseConfig, ) converse_config: Final = AmazonConverseConfig() mapped_params = converse_config.map_openai_params( non_default_params=optional_params, optional_params={}, model=model, drop_params=False, ) return converse_config.transform_request( model=model, messages=messages, optional_params=mapped_params, litellm_params={}, headers={}, ) # --- All other providers: passthrough (OpenAI-compatible models # like openai.gpt-oss-*, qwen, deepseek, etc.) --- return { "messages": messages, **optional_params, } def _resolve_batch_record_model_and_provider( self, record_model: str, target_model: str, ) -> tuple[str, BEDROCK_INVOKE_PROVIDERS_LITERAL | None]: record_provider: Final = self.get_bedrock_invoke_provider(_strip_llm_routing_prefix(record_model)) if record_provider is not None or not target_model: return record_model, record_provider target_provider: Final = self.get_bedrock_invoke_provider(_strip_llm_routing_prefix(target_model)) if target_provider is None: return record_model, record_provider return target_model, target_provider def _transform_openai_jsonl_content_to_bedrock_jsonl_content( self, openai_jsonl_content: Sequence[_OpenAIBatchRecord], target_model: str = "" ) -> list[_BedrockBatchRecord]: """ Transforms OpenAI JSONL content to Bedrock batch format Bedrock batch format: { "recordId": "alphanumeric string", "modelInput": {JSON body} } Example: { "recordId": "CALL0000001", "modelInput": { "anthropic_version": "bedrock-2023-05-31", "max_tokens": 1024, "messages": [ { "role": "user", "content": [{"type": "text", "text": "Hello"}] } ] } } """ import litellm bedrock_jsonl_content: Final = [] for idx, _openai_jsonl_content in enumerate(openai_jsonl_content): # Extract the request body from OpenAI format openai_body = _openai_jsonl_content.get("body", {}) record_model = openai_body.get("model", "") resolved_model = litellm.model_alias_map.get(record_model, record_model) model_for_transform, provider = self._resolve_batch_record_model_and_provider( record_model=resolved_model, target_model=target_model ) # Route to the embedding transformer when the OpenAI batch line # targets /v1/embeddings; every other endpoint shape is normalized # to chat completions first. We branch here (rather than inside # `_map_openai_to_bedrock_params`) so the chat helper keeps its # narrow contract and the embedding helper can evolve independently. record_kind = self._classify_batch_record(_openai_jsonl_content) if record_kind is BedrockBatchRecordKind.EMBEDDING: model_input = self._map_openai_embedding_to_bedrock_params( openai_request_body=openai_body, model=model_for_transform ) else: model_input = self._map_openai_to_bedrock_params( openai_request_body=self._transform_batch_body_to_chat_body(openai_body, record_kind), model=model_for_transform, provider=provider, ) # Create Bedrock batch record record_id = _openai_jsonl_content.get("custom_id", f"CALL{str(idx).zfill(7)}") bedrock_record = {"recordId": record_id, "modelInput": model_input} bedrock_jsonl_content.append(bedrock_record) return bedrock_jsonl_content def transform_create_file_request( self, model: str, create_file_data: CreateFileRequest, optional_params: dict, litellm_params: dict, ) -> bytes | str | dict: """ Transform file request and return a pre-signed request for S3. This keeps the HTTP handler clean by doing all the signing here. """ file_data: Final = create_file_data.get("file") if file_data is None: raise ValueError("file is required") extracted_file_data: Final = extract_file_data(file_data) extracted_file_data_content: Final = extracted_file_data.get("content") if extracted_file_data_content is None: raise ValueError("file content is required") # Get and transform the file content if FilesAPIUtils.is_batch_jsonl_file( create_file_data=create_file_data, extracted_file_data=extracted_file_data, ): ## Transform JSONL content to Bedrock format original_file_content: Final = self._get_content_from_openai_file(extracted_file_data_content) openai_jsonl_content = [json.loads(line) for line in original_file_content.splitlines() if line.strip()] litellm_params_model: Final = litellm_params.get("model") target_model: Final = model or (litellm_params_model if isinstance(litellm_params_model, str) else "") bedrock_jsonl_content = self._transform_openai_jsonl_content_to_bedrock_jsonl_content( openai_jsonl_content, target_model=target_model ) file_content = "\n".join(json.dumps(item) for item in bedrock_jsonl_content) elif isinstance(extracted_file_data_content, bytes): file_content = extracted_file_data_content.decode("utf-8") elif isinstance(extracted_file_data_content, str): file_content = extracted_file_data_content else: raise ValueError("Unsupported file content type") # Get the S3 URL for upload api_base: Final = self.get_complete_file_url( api_base=None, api_key=None, model=model, optional_params=optional_params, litellm_params=litellm_params, data=create_file_data, ) # s3_region_name always wins for S3 operations (same priority as in # get_complete_file_url above). Overwrite aws_region_name unconditionally, # after the deployment-credential merge, so the SigV4 region matches the # URL region, avoiding SignatureDoesNotMatch. merged_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params) s3_region_name: Final = litellm_params.get("s3_region_name") or optional_params.get("s3_region_name") request_params: Final = ( {**merged_params, "aws_region_name": s3_region_name} if s3_region_name else merged_params ) # Sign the request and return a pre-signed request object signed_headers, signed_body = self._sign_s3_request( content=file_content, api_base=api_base, optional_params=request_params, s3_encryption_key_id=resolve_s3_encryption_key_id( litellm_params=litellm_params, optional_params=request_params, ), ) litellm_params["upload_url"] = api_base upload_content_length: Final = len(file_content.encode("utf-8")) litellm_params[UPLOAD_CONTENT_LENGTH_PARAM] = upload_content_length # rebind-ok: same handoff as upload_url # Return a dict that tells the HTTP handler exactly what to do return { "method": "PUT", "url": api_base, "headers": signed_headers, "data": signed_body or file_content, } def _sign_s3_request( self, content: str, api_base: str, optional_params: dict, s3_encryption_key_id: str | None = None, ) -> tuple[dict, str]: """ Sign S3 PUT request using the same proven logic as S3Logger. Reuses the exact pattern from litellm/integrations/s3_v2.py """ try: import hashlib import requests from botocore.auth import S3SigV4Auth from botocore.awsrequest import AWSRequest except ImportError: raise ImportError("Missing boto3 to call bedrock. Run 'pip install boto3'.") # Get AWS credentials using existing methods aws_region_name: Final = self._get_aws_region_name(optional_params=optional_params, model="") credentials: Final = self.get_credentials( aws_access_key_id=optional_params.get("aws_access_key_id"), aws_secret_access_key=optional_params.get("aws_secret_access_key"), aws_session_token=optional_params.get("aws_session_token"), aws_region_name=aws_region_name, aws_session_name=optional_params.get("aws_session_name"), aws_profile_name=optional_params.get("aws_profile_name"), aws_role_name=optional_params.get("aws_role_name"), aws_web_identity_token=optional_params.get("aws_web_identity_token"), aws_sts_endpoint=optional_params.get("aws_sts_endpoint"), ) # Calculate SHA256 hash of the content (REQUIRED for S3) content_hash: Final = hashlib.sha256(content.encode("utf-8")).hexdigest() # Prepare headers with required S3 headers (same as s3_v2.py) sse_headers: Final = ( MappingProxyType( { "x-amz-server-side-encryption": "aws:kms", "x-amz-server-side-encryption-aws-kms-key-id": s3_encryption_key_id, } ) if s3_encryption_key_id else MappingProxyType({}) ) request_headers: Final = MappingProxyType( { "Content-Type": "application/json", # JSONL files are JSON content "x-amz-content-sha256": content_hash, # REQUIRED by S3 "Content-Language": "en", "Cache-Control": "private, immutable, max-age=31536000, s-maxage=0", **sse_headers, } ) # Use requests.Request to prepare the request (same pattern as s3_v2.py) req: Final = requests.Request("PUT", api_base, data=content, headers=request_headers) prepped: Final = req.prepare() # Sign the request with S3 service aws_request: Final = AWSRequest( method=prepped.method, url=prepped.url, data=prepped.body, headers=prepped.headers, ) # Get region name for non-LLM API calls (same as s3_v2.py) signing_region: Final = self.get_aws_region_name_for_non_llm_api_calls(aws_region_name=aws_region_name) S3SigV4Auth(credentials, "s3", signing_region).add_auth(aws_request) # Return signed headers and body signed_body = aws_request.body if isinstance(signed_body, bytes): signed_body = signed_body.decode("utf-8") elif signed_body is None: signed_body = content # Fallback to original content return dict(aws_request.headers), signed_body def _convert_https_url_to_s3_uri(self, https_url: str) -> tuple[str, str]: """ Convert HTTPS S3 URL to s3:// URI format. Args: https_url: HTTPS S3 URL (e.g., "https://s3.us-west-2.amazonaws.com/bucket/key") Returns: Tuple of (s3_uri, filename) Example: Input: "https://s3.us-west-2.amazonaws.com/litellm-proxy/file.jsonl" Output: ("s3://litellm-proxy/file.jsonl", "file.jsonl") """ import re # Match HTTPS S3 URL patterns # Pattern 1: https://s3.region.amazonaws.com/bucket/key # Pattern 2: https://bucket.s3.region.amazonaws.com/key pattern1: Final = r"https://s3\.([^.]+)\.amazonaws\.com/([^/]+)/(.+)" pattern2: Final = r"https://([^.]+)\.s3\.([^.]+)\.amazonaws\.com/(.+)" match1: Final = re.match(pattern1, https_url) match2: Final = re.match(pattern2, https_url) if match1: # Pattern: https://s3.region.amazonaws.com/bucket/key region, bucket, key = match1.groups() key = unquote(key) s3_uri = f"s3://{bucket}/{key}" elif match2: # Pattern: https://bucket.s3.region.amazonaws.com/key bucket, region, key = match2.groups() key = unquote(key) s3_uri = f"s3://{bucket}/{key}" else: # Fallback: try to extract bucket and key from URL path from urllib.parse import urlparse parsed: Final = urlparse(https_url) path_parts: Final = parsed.path.lstrip("/").split("/", 1) if len(path_parts) >= 2: bucket, key = path_parts[0], path_parts[1] key = unquote(key) s3_uri = f"s3://{bucket}/{key}" else: raise ValueError(f"Unable to parse S3 URL: {https_url}") # Extract filename from key filename: Final = key.split("/")[-1] if "/" in key else key return s3_uri, filename def transform_create_file_response( self, model: str | None, raw_response: Response, logging_obj: LiteLLMLoggingObj, litellm_params: dict, ) -> OpenAIFileObject: """ Transform S3 File upload response into OpenAI-style FileObject """ # Use the actual upload URL that was used for the S3 upload upload_url: Final = litellm_params.get("upload_url") file_id: str = "" filename: str = "" if upload_url: # Convert HTTPS S3 URL to s3:// URI format file_id, filename = self._convert_https_url_to_s3_uri(upload_url) return OpenAIFileObject( purpose="batch", # Default purpose for Bedrock files id=file_id, filename=filename, created_at=int(time.time()), # Current timestamp status="uploaded", bytes=_uploaded_object_size(litellm_params=litellm_params, raw_response=raw_response), object="file", ) def get_error_class(self, error_message: str, status_code: int, headers: dict | Headers) -> BaseLLMException: return BedrockError(status_code=status_code, message=error_message, headers=headers) def transform_retrieve_file_request( self, file_id: str, optional_params: dict, litellm_params: dict, ) -> tuple[str, dict]: raise NotImplementedError("BedrockFilesConfig does not support file retrieval") def transform_retrieve_file_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, litellm_params: dict, ) -> OpenAIFileObject: raise NotImplementedError("BedrockFilesConfig does not support file retrieval") def transform_delete_file_request( self, file_id: str, optional_params: dict, litellm_params: dict, ) -> tuple[str, dict]: raise NotImplementedError("BedrockFilesConfig does not support file deletion") def transform_delete_file_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, litellm_params: dict, ) -> FileDeleted: raise NotImplementedError("BedrockFilesConfig does not support file deletion") def transform_list_files_request( self, purpose: str | None, optional_params: dict, litellm_params: dict, ) -> tuple[str, dict]: raise NotImplementedError("BedrockFilesConfig does not support file listing") def transform_list_files_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, litellm_params: dict, ) -> list[OpenAIFileObject]: raise NotImplementedError("BedrockFilesConfig does not support file listing") def transform_file_content_request( self, file_content_request: FileContentRequest, optional_params: Mapping[str, object], litellm_params: MutableMapping[str, object], ) -> tuple[str, dict[str, str]]: """ Build a SigV4-signed S3 GetObject request for a Bedrock batch file. Bedrock batch file ids are `s3://bucket/key` URIs (or unified ids that decode to one); the bucket and key are validated against the server-configured bucket before any request is signed. """ file_id: Final = file_content_request.get("file_id") if not file_id: raise ValueError("file_id is required for Bedrock file content retrieval") s3_uri: Final = extract_s3_uri_from_file_id(file_id) bucket_name, object_key = _validate_file_id_against_configured_buckets( s3_uri=s3_uri, configured_bucket_names=get_configured_s3_bucket_names(litellm_params), allow_legacy_cloud_file_ids=should_allow_legacy_cloud_file_ids(litellm_params), ) # The shared file-content handler passes optional_params={}, so AWS # credentials/region arrive via litellm_params here (unlike the upload # path). s3_region_name wins over aws_region_name, same priority as # get_complete_file_url above. merged_params: Final[dict[str, object]] = {} merged_params.update(litellm_params) merged_params.update(optional_params) request_params: Final = _BedrockS3RequestParams.model_validate(merged_params) region_preference: Final = request_params.s3_region_name or request_params.aws_region_name region_params: Final[dict[str, str | None]] = {"aws_region_name": region_preference} aws_region_name: Final = self._get_aws_region_name(optional_params=region_params, model="") s3_endpoint_url = ( request_params.s3_endpoint_url or f"https://s3.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}" ).rstrip("/") url: Final = f"{s3_endpoint_url}/{bucket_name}/{encode_s3_object_key_for_url(object_key)}" litellm_params[S3_SIGNED_GET_HEADERS_PARAM] = self._sign_s3_get_request( api_base=url, aws_region_name=aws_region_name, request_params=request_params, ) return url, {} def _sign_s3_get_request( self, api_base: str, aws_region_name: str, request_params: _BedrockS3RequestParams, ) -> dict[str, str]: """ SigV4-sign an S3 GetObject request, mirroring `_sign_s3_request` (PUT). """ try: import hashlib from botocore.auth import S3SigV4Auth from botocore.awsrequest import AWSRequest except ImportError: raise ImportError("Missing boto3 to call bedrock. Run 'pip install boto3'.") credentials: Final = self.get_credentials( # any-ok: boto3 Credentials is untyped aws_access_key_id=request_params.aws_access_key_id, aws_secret_access_key=request_params.aws_secret_access_key, aws_session_token=request_params.aws_session_token, aws_region_name=aws_region_name, aws_session_name=request_params.aws_session_name, aws_profile_name=request_params.aws_profile_name, aws_role_name=request_params.aws_role_name, aws_web_identity_token=request_params.aws_web_identity_token, aws_sts_endpoint=request_params.aws_sts_endpoint, ) empty_body_hash: Final = hashlib.sha256(b"").hexdigest() aws_request: Final = AWSRequest( # any-ok: botocore AWSRequest is untyped method="GET", url=api_base, headers={"x-amz-content-sha256": empty_body_hash}, ) auth: Final = S3SigV4Auth(credentials, "s3", aws_region_name) # any-ok: botocore untyped auth.add_auth(aws_request) # any-ok: botocore request mutation is untyped return dict(aws_request.headers) # any-ok: botocore headers are untyped def transform_file_content_response( self, raw_response: httpx.Response, logging_obj: LiteLLMLoggingObj, litellm_params: dict, ) -> HttpxBinaryResponseContent: if raw_response.status_code >= 400: raise BedrockError( status_code=raw_response.status_code, message=raw_response.text, headers=raw_response.headers, ) return HttpxBinaryResponseContent(response=raw_response) class BedrockJsonlFilesTransformation: """ Transforms OpenAI /v1/files/* requests to Bedrock S3 file uploads for batch processing """ def transform_openai_file_content_to_bedrock_file_content( self, openai_file_content: FileTypes | None = None ) -> tuple[str, str]: """ Transforms OpenAI FileContentRequest to Bedrock S3 file format """ if openai_file_content is None: raise ValueError("contents of file are None") # Read the content of the file file_content: Final = self._get_content_from_openai_file(openai_file_content) # Split into lines and parse each line as JSON openai_jsonl_content: Final = [json.loads(line) for line in file_content.splitlines() if line.strip()] bedrock_jsonl_content = self._transform_openai_jsonl_content_to_bedrock_jsonl_content(openai_jsonl_content) bedrock_jsonl_string: Final = "\n".join(json.dumps(item) for item in bedrock_jsonl_content) object_name: Final = self._get_s3_object_name(openai_jsonl_content=openai_jsonl_content) return bedrock_jsonl_string, object_name def _transform_openai_jsonl_content_to_bedrock_jsonl_content( self, openai_jsonl_content: Sequence[_OpenAIBatchRecord] ): """ Delegate to the main BedrockFilesConfig transformation method """ config: Final = BedrockFilesConfig() return config._transform_openai_jsonl_content_to_bedrock_jsonl_content(openai_jsonl_content) def _get_s3_object_name( self, openai_jsonl_content: Sequence[_OpenAIBatchRecord], ) -> str: """ Gets a unique S3 object name for the Bedrock batch processing job named as: litellm-bedrock-files-{model}-{uuid} """ _model = openai_jsonl_content[0].get("body", {}).get("model", "") # Remove bedrock/ prefix if present _model = _model.removeprefix("bedrock/") safe_model: Final = sanitize_cloud_object_component(_model.replace(":", "-"), fallback="model") object_name: Final = f"{BEDROCK_MANAGED_S3_BATCH_PREFIX}{safe_model}-{uuid.uuid4()}.jsonl" return object_name def _get_content_from_openai_file(self, openai_file_content: FileTypes) -> str: """ Helper to extract content from various OpenAI file types and return as string. Handles: - Direct content (str, bytes, IO[bytes]) - Tuple formats: (filename, content, [content_type], [headers]) - PathLike objects """ content: str | bytes = b"" # Extract file content from tuple if necessary if isinstance(openai_file_content, tuple): # Take the second element which is always the file content file_content = openai_file_content[1] else: file_content = openai_file_content # Handle different file content types if isinstance(file_content, str): # String content can be used directly content = file_content elif isinstance(file_content, bytes): # Bytes content can be decoded content = file_content elif isinstance(file_content, PathLike): # PathLike with open(str(file_content), "rb") as f: content = f.read() elif hasattr(file_content, "read"): # IO[bytes] # File-like objects need to be read content = file_content.read() # Ensure content is string if isinstance(content, bytes): content = content.decode("utf-8") return content def transform_s3_bucket_response_to_openai_file_object( self, create_file_data: CreateFileRequest, s3_upload_response: _S3UploadResponse ) -> OpenAIFileObject: """ Transforms S3 Bucket upload file response to OpenAI FileObject """ # S3 response typically contains ETag, key, etc. object_key: Final = s3_upload_response.get("Key", "") bucket_name: Final = s3_upload_response.get("Bucket", "") # Extract filename from object key filename: Final = object_key.split("/")[-1] if "/" in object_key else object_key return OpenAIFileObject( purpose=create_file_data.get("purpose", "batch"), id=f"s3://{bucket_name}/{object_key}", filename=filename, created_at=int(time.time()), # Current timestamp status="uploaded", bytes=s3_upload_response.get("ContentLength", 0), object="file", )