scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-26 01:12:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 98

feat(bedrock): serve the OpenAI models on bedrock-runtime's native Responses API (internal copy of #38489) (#42767)

Browse source 
* feat(bedrock): serve the OpenAI models on bedrock-runtime's native Responses API

AWS serves the OpenAI models on bedrock-runtime through an OpenAI-compatible
surface at /openai/v1/responses, alongside Converse. LiteLLM had no Responses
config for the bedrock provider, so /v1/responses fell back to the Chat
Completions bridge and was translated into Converse. A realistic Codex session
does not survive that translation: its function_call / function_call_output
history becomes Converse toolUse / toolResult blocks with no toolConfig, and
Converse rejects the request outright.

Add a Responses config for that surface, opted into per model from the price-map
supported_endpoints so models without the signal keep the bridge exactly as
before. Auth is Bearer when a Bedrock API key is present, SigV4 otherwise.

Both Bedrock endpoints reject the Codex history item types agent_message,
context_compaction and local_shell_call, so the normalization bedrock_mantle
carried privately moves into a shared module and both providers use it. They are
history items, so they only bite from the second turn onward -- a first-turn
smoke test passes and hides the problem. Verified against bedrock-runtime with
global.openai.gpt-5.6-sol: additional_tools is accepted there (unlike on
bedrock-mantle) while those three types are rejected, so the two endpoints do
not share one validator and each provider opts in explicitly.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* fix(bedrock): build the Responses endpoint from the region's partition suffix

get_complete_url hardcoded amazonaws.com in an f-string, so every non-commercial
partition got the wrong host: cn-north-1 resolved to amazonaws.com instead of
amazonaws.com.cn, and GovCloud/ISO regions were wrong the same way. Defer to
BaseAWSLLM._select_default_endpoint_url, which this config already inherits and
which resolves the suffix per partition.

test_no_fstring_hardcodes_the_commercial_dns_suffix scans the whole tree, so it
caught this even though it is not one of this PR's test files. Register the
config in ENDPOINT_BUILDERS so the cn/GovCloud endpoint sweep covers this
surface from now on rather than only the f-string guard.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* feat(bedrock): opt the gpt-6 family into the native Responses API

* fix(bedrock): drop the Responses tool types bedrock-runtime rejects

Codex sends a web_search tool on every turn. api.openai.com runs that tool
itself, and the Converse bridge dropped it silently, but bedrock-runtime's
native Responses endpoint rejects the whole request with 400 "web search is
not supported for this request". Filter the request's tools down to the
types bedrock-runtime's own validation error names, logging what was dropped,
through a helper shared with the Mantle route, which already did the same.

* fix(bedrock): emulate file_search and collapse custom Responses paths

* fix(bedrock): keep background and remote image inputs working on the native Responses route

* fix(bedrock): inline remote images inside tool outputs on the native Responses route

* fix(bedrock): inline remote computer screenshots on the native Responses route

---------

Co-authored-by: Leonardo Freitas dos Santos <leonardo.freitas.s@outlook.com>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-23 15:12:39 -07:00 • committed by GitHub
parent 170eb7fb95
commit fecc8c8f74
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 1339 additions and 161 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/__init__.py
View file

@ -1872,6 +1872,9 @@ if TYPE_CHECKING:
from .llms.openrouter.responses.transformation import (
OpenRouterResponsesAPIConfig as OpenRouterResponsesAPIConfig,
)
from .llms.bedrock.responses.transformation import (
BedrockOpenAIResponsesConfig as BedrockOpenAIResponsesConfig,
)
from .llms.bedrock_mantle.responses.transformation import (
BedrockMantleResponsesAPIConfig as BedrockMantleResponsesAPIConfig,
)

5
litellm/_lazy_imports_registry.py
View file

@ -245,6 +245,7 @@ LLM_CONFIG_NAMES: Final = (
"PerplexityResponsesConfig",
"DatabricksResponsesAPIConfig",
"OpenRouterResponsesAPIConfig",
"BedrockOpenAIResponsesConfig",
"BedrockMantleResponsesAPIConfig",
"GoogleAIStudioInteractionsConfig",
"VertexAIInteractionsConfig",
@ -921,6 +922,10 @@ _LLM_CONFIGS_IMPORT_MAP: Final = {
"OpenAITextCompletionConfig",
),
"GroqChatConfig": (".llms.groq.chat.transformation", "GroqChatConfig"),
"BedrockOpenAIResponsesConfig": (
".llms.bedrock.responses.transformation",
"BedrockOpenAIResponsesConfig",
),
"BedrockMantleChatConfig": (
".llms.bedrock_mantle.chat.transformation",
"BedrockMantleChatConfig",

154
litellm/llms/base_llm/responses/codex_compat.py Normal file
View file

@ -0,0 +1,154 @@
"""Codex CLI wire-format quirks shared by the Responses API providers that need them.
Codex sends history item types that api.openai.com accepts but other Responses
backends reject with ``400 Invalid 'input': value did not match any expected
variant``. Both Amazon Bedrock endpoints reject them:
- ``bedrock-mantle.{region}.api.aws`` (verified against ``openai.gpt-5.6-sol``)
- ``bedrock-runtime.{region}.amazonaws.com/openai/v1`` (same, verified separately)
They are *history* items, so they only appear from the second turn of a session
onward -- a first-turn request succeeds and hides the problem entirely.
Codex also sends a ``web_search`` tool on every turn. api.openai.com runs that tool
itself; a backend with no server-side tools rejects the whole request over it, so
the same providers drop the tool types their backend does not accept.
Both helpers are pure transforms that report what they rewrote or dropped; callers
do their own logging, so each provider keeps its own wording.
"""
import json
from collections.abc import Mapping, Sequence
from typing import Final
from typing_extensions import ReadOnly, TypedDict
from litellm.types.llms.openai import ResponseInputParam
AGENT_MESSAGE_INPUT_ITEM_TYPE: Final = "agent_message"
CONTEXT_COMPACTION_INPUT_ITEM_TYPE: Final = "context_compaction"
LOCAL_SHELL_CALL_INPUT_ITEM_TYPE: Final = "local_shell_call"
class _RewrittenOutputTextBlock(TypedDict):
type: ReadOnly[str]
text: ReadOnly[str]
class _RewrittenAssistantMessageItem(TypedDict):
type: ReadOnly[str]
role: ReadOnly[str]
content: ReadOnly[tuple[_RewrittenOutputTextBlock, ...]]
class _RewrittenCompactionItem(TypedDict):
type: ReadOnly[str]
encrypted_content: ReadOnly[str]
class _RewrittenFunctionCallItem(TypedDict):
type: ReadOnly[str]
call_id: ReadOnly[str]
name: ReadOnly[str]
arguments: ReadOnly[str]
def _agent_message_text(item: "Mapping[str, object]") -> str:
content: Final = item.get("content")
if not isinstance(content, list):
return ""
return "".join(
str(block.get("text") or block.get("encrypted_content") or "") for block in content if isinstance(block, dict)
)
def _normalize_agent_message_item(item: "Mapping[str, object]") -> "_RewrittenAssistantMessageItem | None":
text: Final = _agent_message_text(item)
if not text:
return None
rewritten: Final[_RewrittenAssistantMessageItem] = {
"type": "message",
"role": "assistant",
"content": ({"type": "output_text", "text": text},),
}
return rewritten
def _normalize_context_compaction_item(item: "Mapping[str, object]") -> "_RewrittenCompactionItem | None":
encrypted_content: Final = item.get("encrypted_content")
if not isinstance(encrypted_content, str) or not encrypted_content:
return None
rewritten: Final[_RewrittenCompactionItem] = {"type": "compaction", "encrypted_content": encrypted_content}
return rewritten
def _normalize_local_shell_call_item(item: "Mapping[str, object]") -> "_RewrittenFunctionCallItem | None":
call_id: Final = item.get("call_id")
if not isinstance(call_id, str) or not call_id:
return None
action: Final = item.get("action")
rewritten: Final[_RewrittenFunctionCallItem] = {
"type": "function_call",
"call_id": call_id,
"name": "local_shell",
"arguments": json.dumps(action) if isinstance(action, dict) else "{}",
}
return rewritten
def _normalize_input_item(item: object) -> "tuple[object, str | None]":
"""Returns (normalized item, or None to drop it; original type when rewritten)."""
if not isinstance(item, dict):
return item, None
item_type: Final = item.get("type")
if item_type == AGENT_MESSAGE_INPUT_ITEM_TYPE:
return _normalize_agent_message_item(item), item_type
if item_type == CONTEXT_COMPACTION_INPUT_ITEM_TYPE:
return _normalize_context_compaction_item(item), item_type
if item_type == LOCAL_SHELL_CALL_INPUT_ITEM_TYPE:
return _normalize_local_shell_call_item(item), item_type
return item, None
def normalize_codex_input_items(
input: "str | ResponseInputParam",
) -> "tuple[str | ResponseInputParam, tuple[str, ...]]":
"""Rewrite the Codex history item types a Responses backend rejects.
``agent_message`` (Codex multi-agent traffic; its ``encrypted_content`` slot
carries the plaintext payload when the model never issued encrypted args)
becomes an assistant message, ``context_compaction`` becomes the ``compaction``
spelling these backends accept, and ``local_shell_call`` becomes the
``function_call`` its recorded ``function_call_output`` already pairs with.
Returns the normalized input and the sorted set of types that were rewritten,
so the caller can log in its own words. Non-list input is returned untouched.
"""
if not isinstance(input, list):
return input, ()
normalized: Final = tuple(_normalize_input_item(item) for item in input)
rewritten_types: Final = tuple(sorted(frozenset(item_type for _, item_type in normalized if item_type is not None)))
kept: Final = [i for i, _ in normalized if i is not None] # mutable-ok: downstream narrows on isinstance(list)
# Codex passthrough items sit outside the OpenAI input union.
return kept, rewritten_types # pyright: ignore[reportReturnType] # see above
def drop_unsupported_tools(
tools: "Sequence[object]", supported_types: "frozenset[str]"
) -> "tuple[tuple[object, ...], tuple[str, ...]]":
"""Keep the tools whose ``type`` the backend accepts; non-dict tools pass through.
Returns the kept tools and the sorted set of dropped types.
"""
kept: Final = tuple(tool for tool in tools if not isinstance(tool, dict) or tool.get("type") in supported_types)
dropped_types: Final = tuple(
sorted(
frozenset(
str(tool.get("type"))
for tool in tools
if isinstance(tool, dict) and tool.get("type") not in supported_types
)
)
)
return kept, dropped_types

16
litellm/llms/base_llm/responses/transformation.py
View file

@ -130,6 +130,22 @@ class BaseResponsesAPIConfig(ABC):
) -> dict:
pass
async def async_transform_responses_api_request(
self,
model: str,
input: str | ResponseInputParam,
response_api_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict,
) -> dict:
return self.transform_responses_api_request(
model=model,
input=input,
response_api_optional_request_params=response_api_optional_request_params,
litellm_params=litellm_params,
headers=headers,
)
@abstractmethod
def transform_response_api_response(
self,

22
litellm/llms/bedrock/common_utils.py
View file

@ -827,6 +827,28 @@ def _mantle_api_base_from_env() -> str | None:
return next((base[: -len(suffix)] for suffix in _MANTLE_OPENAI_BASE_SUFFIXES if base.endswith(suffix)), base)
def bedrock_supports_openai_responses(model: str | None, model_cost: Mapping[str, object]) -> bool:
"""Whether a Bedrock model is served by bedrock-runtime's OpenAI Responses surface.
Purely data-driven from the model's price-map capability signal -- ``/v1/responses``
in ``supported_endpoints`` -- and overridable via ``register_model`` and proxy
``model_info``, so onboarding a model is a JSON change, never a code change.
There is deliberately no model-name match: AWS exposes this surface per model,
not per family, and the two Bedrock endpoints do not agree with each other
(bedrock-runtime accepts Codex's ``additional_tools`` items where
bedrock-mantle rejects them), so a name-shaped gate would be wrong.
A model absent from ``model_cost`` has no signal and returns False, leaving the
chat-completions bridge in place exactly as before.
"""
if not model:
return False
candidates: Final = (model_cost.get(key) for key in (model, f"bedrock/{model}"))
return any(
isinstance(entry, Mapping) and "/v1/responses" in (entry.get("supported_endpoints") or ())
for entry in candidates
)
def build_mantle_messages_url(
api_base: str | None,
aws_bedrock_runtime_endpoint: str | None,

338
litellm/llms/bedrock/responses/transformation.py Normal file
View file

@ -0,0 +1,338 @@
"""Amazon Bedrock Runtime - native OpenAI Responses API.
AWS serves the OpenAI models on ``bedrock-runtime`` through an OpenAI-compatible
surface at ``https://bedrock-runtime.{region}.{dns_suffix}/openai/v1/responses``,
alongside Converse. Without this config the ``bedrock`` provider has no Responses
config at all, so ``/v1/responses`` falls back to the Chat Completions bridge and
the request is translated into Converse, which rejects Responses-only parameters
such as ``prompt_cache_key`` with a 400 and never sees reasoning items.
Payloads and SSE follow the OpenAI Responses spec, so this inherits
OpenAIResponsesAPIConfig and overrides only the endpoint URL, authentication, the
Codex history-item normalization the endpoint requires, and the tool filter below.
Tools: bedrock-runtime runs no server-side tools, so it rejects Codex's default
``web_search`` tool with "web search is not supported for this request". The
Converse bridge dropped that tool silently (Converse has no web search either),
so this config drops every tool type the endpoint rejects the same way. The
supported set is the one bedrock-runtime's own validation error names.
Parity with the Converse bridge on what it used to accept: ``background`` never
reached Converse (the bridge answered synchronously), while bedrock-runtime rejects
it with "The background parameter is not supported.", so it is dropped here. The
bridge also downloaded ``input_image`` http(s) URLs for Converse, while
bedrock-runtime only accepts ``data:`` and ``s3://`` image URLs, so remote image
URLs are fetched and inlined as data URIs before the request is signed.
Auth: Bearer token (litellm_params.api_key or the standard AWS_BEARER_TOKEN_BEDROCK)
when present; otherwise AWS SigV4 (service "bedrock") over the standard credential
chain, signed via BaseAWSLLM._sign_request once the body is final.
Model IDs: bedrock-runtime serves these models only through a cross-Region
inference profile, so the model is named ``us.openai.gpt-5.6-sol`` or
``global.openai.gpt-5.6-sol``; there is no in-Region form.
"""
import asyncio
from collections.abc import Awaitable, Callable, Mapping
from types import MappingProxyType
from typing import Final
import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.prompt_templates.image_handling import (
async_convert_url_to_base64,
convert_url_to_base64,
)
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.responses.codex_compat import drop_unsupported_tools, normalize_codex_input_items
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import (
BedrockError,
bedrock_supports_openai_responses,
)
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import ResponseInputParam, ResponsesAPIOptionalRequestParams
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
BEDROCK_RUNTIME_OPENAI_RESPONSES_PATH: Final = "/openai/v1/responses"
BEDROCK_RUNTIME_OPENAI_BASE_SUFFIXES: Final = (
"/openai/v1/responses",
"/v1/responses",
"/responses",
"/openai/v1",
"/v1",
)
BEDROCK_RUNTIME_SUPPORTED_RESPONSE_TOOL_TYPES: Final = frozenset(
{"function", "mcp", "custom", "apply_patch", "namespace", "tool_search", "computer"}
)
BEDROCK_RUNTIME_UNSUPPORTED_RESPONSE_PARAMS: Final = frozenset({"background"})
REMOTE_IMAGE_URL_SCHEMES: Final = ("http://", "https://")
IMAGE_BLOCK_KEYS: Final = ("content", "output")
IMAGE_BLOCK_TYPES: Final = frozenset({"input_image", "computer_screenshot"})
def resolve_bedrock_bearer_token(api_key: str | None) -> str | None:
return api_key or get_secret_str("AWS_BEARER_TOKEN_BEDROCK")
def _remote_image_url(block: object) -> str | None:
if not isinstance(block, dict) or block.get("type") not in IMAGE_BLOCK_TYPES:
return None
image_url: Final = block.get("image_url")
if not isinstance(image_url, str) or not image_url.startswith(REMOTE_IMAGE_URL_SCHEMES):
return None
return image_url
def _blocks_under(value: object) -> "tuple[object, ...]":
if isinstance(value, list):
return tuple(value)
if isinstance(value, dict):
return (value,)
return ()
def _image_blocks(item: object) -> "tuple[object, ...]":
"""The blocks of ``item`` that can carry an image: its content and tool output lists, or a screenshot output dict."""
if not isinstance(item, dict):
return ()
return tuple(block for key in IMAGE_BLOCK_KEYS for block in _blocks_under(item.get(key)))
def collect_remote_image_urls(input: "str | ResponseInputParam") -> "tuple[str, ...]":
"""The distinct http(s) image URLs in message content, tool output lists, and computer screenshots, in first-seen order."""
if not isinstance(input, list):
return ()
return tuple(
dict.fromkeys(
url for item in input for block in _image_blocks(item) if (url := _remote_image_url(block)) is not None
)
)
def _inline_block(block: object, inlined: "Mapping[str, str]") -> object:
url: Final = _remote_image_url(block)
if url is None or not isinstance(block, dict):
return block
return {**block, "image_url": inlined[url]} # mutable-ok: outgoing JSON request item
def _inline_value(value: object, inlined: "Mapping[str, str]") -> object:
if isinstance(value, list):
return [_inline_block(block, inlined) for block in value] # mutable-ok: outgoing JSON request item
return _inline_block(value, inlined)
def _inline_item(item: object, inlined: "Mapping[str, str]") -> object:
if not isinstance(item, dict):
return item
inlined_fields: Final = { # mutable-ok: outgoing JSON request item
key: _inline_value(item[key], inlined) for key in IMAGE_BLOCK_KEYS if isinstance(item.get(key), (list, dict))
}
if not inlined_fields:
return item
return {**item, **inlined_fields} # mutable-ok: same
def inline_remote_image_urls(
input: "str | ResponseInputParam", inlined: "Mapping[str, str]"
) -> "str | ResponseInputParam":
"""``input`` with every http(s) image URL replaced by its entry in ``inlined``."""
if not isinstance(input, list) or not inlined:
return input
items: Final = [_inline_item(item, inlined) for item in input] # mutable-ok: downstream narrows on isinstance(list)
return items # pyright: ignore[reportReturnType] # items keep the caller's input union
class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
"""Responses API config for the OpenAI models on the bedrock-runtime endpoint."""
def __init__(
self,
fetch_image: "Callable[[str], str]" = convert_url_to_base64,
async_fetch_image: "Callable[[str], Awaitable[str]]" = async_convert_url_to_base64,
) -> None:
super().__init__()
self.fetch_image = fetch_image
self.async_fetch_image = async_fetch_image
@classmethod
def for_model(cls, model: str | None) -> "BedrockOpenAIResponsesConfig | None":
"""This config when ``model`` is served on the OpenAI Responses surface, else ``None``.
The capability decision lives here rather than in the shared dispatch so that
onboarding a model, or changing how the signal is read, stays inside the
Bedrock adapter. ``None`` leaves the caller's existing behaviour untouched --
chat-only Bedrock models keep the Chat Completions bridge.
"""
if not bedrock_supports_openai_responses(model, litellm.model_cost):
return None
return cls()
@property
def custom_llm_provider(self) -> LlmProviders:
return LlmProviders.BEDROCK
def get_error_class(
self, error_message: str, status_code: int, headers: dict[str, object] | httpx.Headers
) -> BaseLLMException:
# The OpenAI base builds a blank response, dropping x-amzn-RequestId.
return BedrockError(status_code=status_code, message=error_message, headers=headers)
def get_complete_url(
self,
api_base: str | None,
litellm_params: dict, # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
) -> str:
region: Final = self._get_aws_region_name(optional_params=litellm_params, model=None)
override: Final = (
api_base
or litellm_params.get("aws_bedrock_runtime_endpoint")
or get_secret_str("AWS_BEDROCK_RUNTIME_ENDPOINT")
)
# Partition-aware: bedrock-runtime is amazonaws.com.cn in China, and other
# suffixes in GovCloud/ISO, so defer to the shared endpoint builder.
host: Final = (
override or self._select_default_endpoint_url(endpoint_type="runtime", aws_region_name=region)
).rstrip("/")
base: Final = next(
(host[: -len(suffix)] for suffix in BEDROCK_RUNTIME_OPENAI_BASE_SUFFIXES if host.endswith(suffix)),
host,
)
return f"{base}{BEDROCK_RUNTIME_OPENAI_RESPONSES_PATH}"
def supports_native_file_search(self) -> bool:
return False
def validate_environment(
self,
headers: dict, # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
model: str,
litellm_params: GenericLiteLLMParams | None,
) -> dict: # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
api_key: Final = litellm_params.api_key if litellm_params is not None else None
bearer: Final = resolve_bedrock_bearer_token(api_key)
if not bearer:
return headers
return {**headers, "Authorization": f"Bearer {bearer}"} # mutable-ok: dict return per the contract
def sign_request(
self,
headers: dict, # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
optional_params: dict, # mutable-ok: same
request_data: dict, # mutable-ok: same
api_base: str,
api_key: str | None = None,
model: str | None = None,
stream: bool | None = None,
fake_stream: bool | None = None,
) -> "tuple[dict, bytes | None]": # mutable-ok: signature fixed by the override contract
if resolve_bedrock_bearer_token(api_key):
# Bedrock API keys are Bearer credentials; SigV4 on top would be wrong.
return headers, None
return self._sign_request(
service_name="bedrock",
headers=headers,
optional_params=optional_params,
request_data=request_data,
api_base=api_base,
model=model,
stream=stream,
fake_stream=fake_stream,
)
def map_openai_params(
self,
response_api_optional_params: ResponsesAPIOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict: # mutable-ok: signature fixed by the override contract
mapped: Final = super().map_openai_params(
response_api_optional_params=response_api_optional_params, model=model, drop_params=drop_params
)
unsupported: Final = tuple(sorted(BEDROCK_RUNTIME_UNSUPPORTED_RESPONSE_PARAMS & mapped.keys()))
if unsupported:
verbose_logger.warning(
"Bedrock Runtime Responses API: dropping unsupported parameter(s) %s that the endpoint rejects.",
unsupported,
)
params: Final = { # mutable-ok: outgoing JSON request params
key: value for key, value in mapped.items() if key not in unsupported
}
tools: Final = params.get("tools")
if not isinstance(tools, list):
return params
kept, dropped_types = drop_unsupported_tools(tools, BEDROCK_RUNTIME_SUPPORTED_RESPONSE_TOOL_TYPES)
if not dropped_types:
return params
verbose_logger.warning(
"Bedrock Runtime Responses API: dropping unsupported tool type(s) %s (supported: %s).",
list(dropped_types),
sorted(BEDROCK_RUNTIME_SUPPORTED_RESPONSE_TOOL_TYPES),
)
without_tools: Final = {key: value for key, value in params.items() if key != "tools"}
if not kept:
return without_tools
return {**without_tools, "tools": list(kept)}
def transform_responses_api_request(
self,
model: str,
input: "str | ResponseInputParam",
response_api_optional_request_params: dict, # mutable-ok: signature fixed by the override contract
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: same
) -> dict: # mutable-ok: same
inlined: Final = MappingProxyType({url: self.fetch_image(url) for url in collect_remote_image_urls(input)})
return self._transform_inlined_request(
model=model,
input=inline_remote_image_urls(input, inlined),
response_api_optional_request_params=response_api_optional_request_params,
litellm_params=litellm_params,
headers=headers,
)
async def async_transform_responses_api_request(
self,
model: str,
input: "str | ResponseInputParam",
response_api_optional_request_params: dict, # mutable-ok: signature fixed by the override contract
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: same
) -> dict: # mutable-ok: same
remote_urls: Final = collect_remote_image_urls(input)
data_uris: Final = await asyncio.gather(*(self.async_fetch_image(url) for url in remote_urls))
return self._transform_inlined_request(
model=model,
input=inline_remote_image_urls(input, MappingProxyType(dict(zip(remote_urls, data_uris, strict=True)))),
response_api_optional_request_params=response_api_optional_request_params,
litellm_params=litellm_params,
headers=headers,
)
def _transform_inlined_request(
self,
model: str,
input: "str | ResponseInputParam",
response_api_optional_request_params: dict, # mutable-ok: signature fixed by the override contract
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: same
) -> dict: # mutable-ok: same
normalized_input, rewritten_types = normalize_codex_input_items(input)
if rewritten_types:
verbose_logger.warning(
"Bedrock Runtime Responses API: rewrote Codex input item type(s) %s that the endpoint rejects.",
rewritten_types,
)
return super().transform_responses_api_request(
model=model,
input=normalized_input,
response_api_optional_request_params=response_api_optional_request_params,
litellm_params=litellm_params,
headers=headers,
)

140
litellm/llms/bedrock_mantle/responses/transformation.py
View file

@ -15,16 +15,15 @@ role / access key / profile / web identity), signed via the shared
BaseAWSLLM._sign_request after the request body is finalized.
"""
import json
from collections.abc import Mapping, Sequence
from typing import Final, cast # noqa: TID251 # map_openai_params returns the filtered params as a bare dict
import httpx
from typing_extensions import ReadOnly, TypedDict
import litellm
from litellm._logging import verbose_logger
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.responses.codex_compat import drop_unsupported_tools, normalize_codex_input_items
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.bedrock_mantle.common_utils import (
@ -59,33 +58,6 @@ _BEDROCK_MANTLE_SUPPORTED_RESPONSE_TOOL_TYPES: Final = frozenset(
_BEDROCK_MANTLE_SUPPORTED_SERVICE_TIERS: Final = frozenset({"auto", "default"})
_BEDROCK_MANTLE_OPENAI_PATH_SUPPORTED_REASONING_SUMMARIES: Final = frozenset({"auto"})
_CODEX_AGENT_MESSAGE_INPUT_ITEM_TYPE: Final = "agent_message"
_CODEX_CONTEXT_COMPACTION_INPUT_ITEM_TYPE: Final = "context_compaction"
_CODEX_LOCAL_SHELL_CALL_INPUT_ITEM_TYPE: Final = "local_shell_call"
class _RewrittenOutputTextBlock(TypedDict):
type: ReadOnly[str]
text: ReadOnly[str]
class _RewrittenAssistantMessageItem(TypedDict):
type: ReadOnly[str]
role: ReadOnly[str]
content: ReadOnly[tuple[_RewrittenOutputTextBlock, ...]]
class _RewrittenCompactionItem(TypedDict):
type: ReadOnly[str]
encrypted_content: ReadOnly[str]
class _RewrittenFunctionCallItem(TypedDict):
type: ReadOnly[str]
call_id: ReadOnly[str]
name: ReadOnly[str]
arguments: ReadOnly[str]
class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPIConfig):
def __init__(
@ -144,26 +116,14 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
@staticmethod
def _filter_unsupported_tools(tools: "Sequence[object]") -> "list[object]":
"""Keep only tool types Mantle's Responses API accepts."""
kept: Final[list[object]] = []
dropped_types: Final[list[str]] = []
for tool in tools:
if not isinstance(tool, dict):
kept.append(tool)
continue
tool_type = tool.get("type")
if tool_type in _BEDROCK_MANTLE_SUPPORTED_RESPONSE_TOOL_TYPES:
kept.append(tool)
else:
dropped_types.append(str(tool_type))
kept, dropped_types = drop_unsupported_tools(tools, _BEDROCK_MANTLE_SUPPORTED_RESPONSE_TOOL_TYPES)
if dropped_types:
verbose_logger.warning(
"Bedrock Mantle Responses API: dropping unsupported tool type(s) %s (supported: %s).",
sorted(set(dropped_types)),
list(dropped_types),
sorted(_BEDROCK_MANTLE_SUPPORTED_RESPONSE_TOOL_TYPES),
)
return kept
return list(kept)
@staticmethod
def _handle_unsupported_service_tier(params: dict, drop_params: bool) -> dict:
@ -236,7 +196,12 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
"ResponsesAPIOptionalRequestParams", response_api_optional_request_params
)
hoisted: Final = hoist_additional_tools(input, params.get("tools"))
normalized_input: Final = self._normalize_codex_input_items(hoisted.input)
normalized_input, rewritten_types = normalize_codex_input_items(hoisted.input)
if rewritten_types:
verbose_logger.warning(
"Bedrock Mantle Responses API: rewrote Codex input item type(s) %s that Mantle rejects.",
list(rewritten_types),
)
request_params: Final = (
self._params_with_hoisted_tools(params, hoisted)
if hoisted.hoisted
@ -259,91 +224,6 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
return {**params, "tools": supported_tools}
return {key: value for key, value in params.items() if key != "tools"}
@staticmethod
def _agent_message_text(item: "Mapping[str, object]") -> str:
content: Final = item.get("content")
if not isinstance(content, list):
return ""
return "".join(
str(block.get("text") or block.get("encrypted_content") or "")
for block in content
if isinstance(block, dict)
)
@classmethod
def _normalize_agent_message_item(cls, item: "Mapping[str, object]") -> "_RewrittenAssistantMessageItem | None":
text: Final = cls._agent_message_text(item)
if not text:
return None
rewritten: Final[_RewrittenAssistantMessageItem] = {
"type": "message",
"role": "assistant",
"content": ({"type": "output_text", "text": text},),
}
return rewritten
@staticmethod
def _normalize_context_compaction_item(item: "Mapping[str, object]") -> "_RewrittenCompactionItem | None":
encrypted_content: Final = item.get("encrypted_content")
if not isinstance(encrypted_content, str) or not encrypted_content:
return None
rewritten: Final[_RewrittenCompactionItem] = {"type": "compaction", "encrypted_content": encrypted_content}
return rewritten
@staticmethod
def _normalize_local_shell_call_item(item: "Mapping[str, object]") -> "_RewrittenFunctionCallItem | None":
call_id: Final = item.get("call_id")
if not isinstance(call_id, str) or not call_id:
return None
action: Final = item.get("action")
rewritten: Final[_RewrittenFunctionCallItem] = {
"type": "function_call",
"call_id": call_id,
"name": "local_shell",
"arguments": json.dumps(action) if isinstance(action, dict) else "{}",
}
return rewritten
@classmethod
def _normalize_codex_input_item(cls, item: object) -> "tuple[object, str | None]":
"""Returns (normalized item or None to drop it, original type when rewritten)."""
if not isinstance(item, dict):
return item, None
item_type: Final = item.get("type")
if item_type == _CODEX_AGENT_MESSAGE_INPUT_ITEM_TYPE:
return cls._normalize_agent_message_item(item), item_type
if item_type == _CODEX_CONTEXT_COMPACTION_INPUT_ITEM_TYPE:
return cls._normalize_context_compaction_item(item), item_type
if item_type == _CODEX_LOCAL_SHELL_CALL_INPUT_ITEM_TYPE:
return cls._normalize_local_shell_call_item(item), item_type
return item, None
@classmethod
def _normalize_codex_input_items(
cls,
input: "str | ResponseInputParam",
) -> "str | ResponseInputParam":
"""Rewrite Codex history item types Mantle rejects with 400 "Invalid
'input': value did not match any expected variant" into supported
equivalents. `agent_message` (Codex multi-agent traffic; its
encrypted_content slot carries the plaintext payload when the model
never issued encrypted args) becomes an assistant message,
`context_compaction` becomes the `compaction` spelling Mantle accepts,
and `local_shell_call` becomes the function_call its recorded
function_call_output already pairs with.
"""
if not isinstance(input, list):
return input
normalized: Final = tuple(cls._normalize_codex_input_item(item) for item in input)
rewritten_types: Final = sorted(frozenset(item_type for _, item_type in normalized if item_type is not None))
if rewritten_types:
verbose_logger.warning(
"Bedrock Mantle Responses API: rewrote Codex input item type(s) %s that Mantle rejects.",
rewritten_types,
)
kept: Final = [item for item, _ in normalized if item is not None] # mutable-ok: ResponseInputParam is a list
return kept # pyright: ignore[reportReturnType] # Codex passthrough items sit outside the OpenAI input union
@staticmethod
def _model_map_lookup_name(model: str) -> str:
return model.split("/")[-1].removeprefix("openai.")

2
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -2881,7 +2881,7 @@ class BaseLLMHTTPHandler:
litellm_params=dict(litellm_params),
)
data = responses_api_provider_config.transform_responses_api_request(
data = await responses_api_provider_config.async_transform_responses_api_request(
model=model,
input=input,
response_api_optional_request_params=response_api_optional_request_params,

60
litellm/model_prices_and_context_window_backup.json
View file

@ -55923,7 +55923,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-5.6-sol": {
"input_cost_per_token": 4e-06,
@ -55954,7 +55957,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-5.6-terra": {
"input_cost_per_token": 2.2e-06,
@ -55985,7 +55991,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-5.6-terra": {
"input_cost_per_token": 2e-06,
@ -56016,7 +56025,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-5.6-luna": {
"input_cost_per_token": 2.2e-07,
@ -56047,7 +56059,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-5.6-luna": {
"input_cost_per_token": 2e-07,
@ -56078,7 +56093,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"bedrock_mantle/openai.gpt-6-astra": {
"input_cost_per_token": 1.1e-05,
@ -56224,7 +56242,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-6-sol": {
"input_cost_per_token": 2.2e-06,
@ -56256,7 +56277,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-6-luna": {
"input_cost_per_token": 1.1e-07,
@ -56288,7 +56312,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-6-astra": {
"input_cost_per_token": 1e-05,
@ -56320,7 +56347,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"openai.gpt-6-sol": {
"input_cost_per_token": 2e-06,
@ -56384,7 +56414,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"openai.gpt-6-luna": {
"input_cost_per_token": 1e-07,
@ -56448,7 +56481,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"bedrock_mantle/openai.gpt-5.5": {
"input_cost_per_token": 5.5e-06,

5
litellm/utils.py
View file

@ -9074,6 +9074,11 @@ class ProviderConfigManager:
return litellm.FireworksAIResponsesAPIConfig()
elif litellm.LlmProviders.EDENAI == provider:
return litellm.EdenAIResponsesAPIConfig()
elif litellm.LlmProviders.BEDROCK == provider:
# bedrock-runtime serves the OpenAI models on an OpenAI-compatible surface
# (/openai/v1/responses) alongside Converse. The adapter decides whether a
# given model is on it; None keeps the chat-completions bridge.
return litellm.BedrockOpenAIResponsesConfig.for_model(model)
elif litellm.LlmProviders.BEDROCK_MANTLE == provider:
# Both decisions are data-driven from the model's price-map entry, with
# no model-name logic. Capability (can it serve Responses?) comes from

60
model_prices_and_context_window.json
View file

@ -55923,7 +55923,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-5.6-sol": {
"input_cost_per_token": 4e-06,
@ -55954,7 +55957,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-5.6-terra": {
"input_cost_per_token": 2.2e-06,
@ -55985,7 +55991,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-5.6-terra": {
"input_cost_per_token": 2e-06,
@ -56016,7 +56025,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-5.6-luna": {
"input_cost_per_token": 2.2e-07,
@ -56047,7 +56059,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-5.6-luna": {
"input_cost_per_token": 2e-07,
@ -56078,7 +56093,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"supports_sampling_params": false
"supports_sampling_params": false,
"supported_endpoints": [
"/v1/responses"
]
},
"bedrock_mantle/openai.gpt-6-astra": {
"input_cost_per_token": 1.1e-05,
@ -56224,7 +56242,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-6-sol": {
"input_cost_per_token": 2.2e-06,
@ -56256,7 +56277,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"us.openai.gpt-6-luna": {
"input_cost_per_token": 1.1e-07,
@ -56288,7 +56312,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"global.openai.gpt-6-astra": {
"input_cost_per_token": 1e-05,
@ -56320,7 +56347,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"openai.gpt-6-sol": {
"input_cost_per_token": 2e-06,
@ -56384,7 +56414,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"openai.gpt-6-luna": {
"input_cost_per_token": 1e-07,
@ -56448,7 +56481,10 @@
"supports_reasoning": true,
"supports_xhigh_reasoning_effort": true,
"supports_vision": true,
"source": "https://aws.amazon.com/bedrock/pricing/"
"source": "https://aws.amazon.com/bedrock/pricing/",
"supported_endpoints": [
"/v1/responses"
]
},
"bedrock_mantle/openai.gpt-5.5": {
"input_cost_per_token": 5.5e-06,

5
tests/test_litellm/litellm_core_utils/test_aws_partition.py
View file

@ -21,6 +21,7 @@ from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.batches.transformation import BedrockBatchesConfig
from litellm.llms.bedrock.chat.agentcore.transformation import AmazonAgentCoreConfig
from litellm.llms.bedrock.common_utils import init_bedrock_client
from litellm.llms.bedrock.responses.transformation import BedrockOpenAIResponsesConfig
from litellm.llms.sagemaker.chat.transformation import SagemakerChatConfig
@ -152,6 +153,10 @@ ENDPOINT_BUILDERS: Final = {
litellm_params={},
stream=True,
),
"bedrock_openai_responses": lambda region: BedrockOpenAIResponsesConfig().get_complete_url(
api_base=None,
litellm_params={"aws_region_name": region},
),
"s3_object_url": _s3_object_url,
}

116
tests/test_litellm/llms/base_llm/responses/test_codex_compat.py Normal file
View file

@ -0,0 +1,116 @@
"""Shared Codex wire-format normalization.
Both Bedrock endpoints reject the Codex *history* item types with
``400 Invalid 'input': value did not match any expected variant``. They are history
items, so they only appear from the second turn of a session onward — a first-turn
smoke test passes and hides the problem entirely.
"""
import json
import pytest
from litellm.llms.base_llm.responses.codex_compat import normalize_codex_input_items
USER = {"role": "user", "content": "hi"}
class TestAgentMessage:
def test_becomes_an_assistant_message(self):
item = {
"type": "agent_message",
"role": "assistant",
"content": [{"type": "output_text", "text": "prior turn"}],
}
out, types = normalize_codex_input_items([item, USER])
assert types == ("agent_message",)
assert out[0] == {
"type": "message",
"role": "assistant",
"content": ({"type": "output_text", "text": "prior turn"},),
}
def test_encrypted_content_slot_is_used_as_text(self):
"""Codex puts the plaintext payload there when the model issued no encrypted args."""
item = {"type": "agent_message", "content": [{"encrypted_content": "plain"}]}
out, _ = normalize_codex_input_items([item, USER])
assert out[0]["content"] == ({"type": "output_text", "text": "plain"},)
def test_non_list_content_yields_no_text_and_drops_the_item(self):
out, types = normalize_codex_input_items([{"type": "agent_message", "content": "not a list"}, USER])
assert out == [USER]
assert types == ("agent_message",)
def test_textless_item_is_dropped(self):
out, types = normalize_codex_input_items([{"type": "agent_message", "content": []}, USER])
assert out == [USER]
assert types == ("agent_message",)
class TestContextCompaction:
def test_becomes_compaction(self):
out, types = normalize_codex_input_items([{"type": "context_compaction", "encrypted_content": "abc"}, USER])
assert out[0] == {"type": "compaction", "encrypted_content": "abc"}
assert types == ("context_compaction",)
@pytest.mark.parametrize("bad", [{}, {"encrypted_content": ""}, {"encrypted_content": 7}])
def test_without_usable_content_is_dropped(self, bad):
out, _ = normalize_codex_input_items([{"type": "context_compaction", **bad}, USER])
assert out == [USER]
class TestLocalShellCall:
def test_becomes_the_function_call_its_output_pairs_with(self):
out, types = normalize_codex_input_items(
[{"type": "local_shell_call", "call_id": "c1", "action": {"command": ["ls"]}}, USER]
)
assert out[0] == {
"type": "function_call",
"call_id": "c1",
"name": "local_shell",
"arguments": json.dumps({"command": ["ls"]}),
}
assert types == ("local_shell_call",)
def test_missing_action_yields_empty_arguments(self):
out, _ = normalize_codex_input_items([{"type": "local_shell_call", "call_id": "c1"}, USER])
assert out[0]["arguments"] == "{}"
def test_without_call_id_is_dropped(self):
out, _ = normalize_codex_input_items([{"type": "local_shell_call"}, USER])
assert out == [USER]
class TestPassthroughAndShape:
def test_string_input_untouched(self):
assert normalize_codex_input_items("just a prompt") == ("just a prompt", ())
def test_unrelated_items_untouched_and_no_types_reported(self):
items = [USER, {"type": "message", "role": "assistant", "content": []}]
out, types = normalize_codex_input_items(items)
assert out == items
assert types == ()
def test_non_mapping_entries_pass_through_except_a_literal_none(self):
"""A literal ``None`` is indistinguishable from "drop this item" in the
per-item return protocol, so it is dropped. Other non-mapping entries pass
through untouched. This matches the behaviour before the normalizer moved
out of the bedrock_mantle config."""
out, types = normalize_codex_input_items(["a string", 42, None, USER])
assert out == ["a string", 42, USER]
assert types == ()
def test_types_are_sorted_and_deduplicated(self):
items = [
{"type": "local_shell_call", "call_id": "c1"},
{"type": "agent_message", "content": [{"text": "x"}]},
{"type": "local_shell_call", "call_id": "c2"},
]
_, types = normalize_codex_input_items(items)
assert types == ("agent_message", "local_shell_call")
def test_returns_a_list_not_a_tuple(self):
"""The input->messages conversion downstream narrows on isinstance(input, list);
a tuple silently yields zero messages and the provider rejects the request."""
out, _ = normalize_codex_input_items([{"type": "agent_message", "content": [{"text": "x"}]}, USER])
assert isinstance(out, list)

35
tests/test_litellm/llms/base_llm/responses/test_transformation.py Normal file
View file

@ -0,0 +1,35 @@
"""The shared Responses API config contract."""
import pytest
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.types.router import GenericLiteLLMParams
@pytest.mark.asyncio
async def test_default_async_transform_delegates_to_the_sync_transform():
"""A config that overrides only the sync transform gets the same request from the async hook,
so the async handler can always await the hook."""
cfg = OpenAIResponsesAPIConfig()
input_with_cache_marker = [
{
"role": "user",
"content": [{"type": "input_text", "text": "hi", "cache_control": {"type": "ephemeral"}}],
}
]
sync_body = cfg.transform_responses_api_request(
model="gpt-5",
input=input_with_cache_marker,
response_api_optional_request_params={"max_output_tokens": 64},
litellm_params=GenericLiteLLMParams(),
headers={},
)
async_body = await cfg.async_transform_responses_api_request(
model="gpt-5",
input=input_with_cache_marker,
response_api_optional_request_params={"max_output_tokens": 64},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert async_body == sync_body
assert "cache_control" not in async_body["input"][0]["content"][0]

492
tests/test_litellm/llms/bedrock/responses/test_bedrock_openai_responses.py Normal file
View file

@ -0,0 +1,492 @@
"""Native OpenAI Responses API on the bedrock-runtime endpoint.
Without this config the bedrock provider has no Responses config, so /v1/responses
falls back to the Chat Completions bridge and rides Converse.
"""
import json
import logging
from importlib.resources import files
from unittest.mock import patch
import pytest
import litellm
from litellm.llms.bedrock.common_utils import bedrock_supports_openai_responses
from litellm.llms.bedrock.responses.transformation import BedrockOpenAIResponsesConfig
from litellm.responses.file_search.emulated_handler import should_use_emulated_file_search
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
MODEL = "global.openai.gpt-5.6-sol"
def _cfg():
return BedrockOpenAIResponsesConfig()
class TestCompleteURL:
def test_default_host_and_path(self):
url = _cfg().get_complete_url(None, {"aws_region_name": "us-east-1"})
assert url == "https://bedrock-runtime.us-east-1.amazonaws.com/openai/v1/responses"
def test_region_is_honoured(self):
url = _cfg().get_complete_url(None, {"aws_region_name": "eu-west-1"})
assert url == "https://bedrock-runtime.eu-west-1.amazonaws.com/openai/v1/responses"
@pytest.mark.parametrize(
"api_base",
[
"https://proxy.example.com",
"https://proxy.example.com/",
"https://proxy.example.com/openai/v1",
"https://proxy.example.com/openai/v1/responses",
"https://proxy.example.com/v1",
"https://proxy.example.com/v1/responses",
"https://proxy.example.com/responses",
],
)
def test_custom_host_is_preserved_and_path_never_doubles(self, api_base):
url = _cfg().get_complete_url(api_base, {"aws_region_name": "us-east-1"})
assert url == "https://proxy.example.com/openai/v1/responses"
def test_runtime_endpoint_param_is_honoured(self):
url = _cfg().get_complete_url(
None, {"aws_region_name": "us-east-1", "aws_bedrock_runtime_endpoint": "https://vpce.example.com"}
)
assert url == "https://vpce.example.com/openai/v1/responses"
class TestAuth:
def test_bearer_token_is_used_when_present(self):
headers = _cfg().validate_environment({}, MODEL, GenericLiteLLMParams(api_key="sk-bedrock"))
assert headers["Authorization"] == "Bearer sk-bedrock"
def test_no_authorization_header_without_a_token(self, monkeypatch):
monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False)
headers = _cfg().validate_environment({}, MODEL, GenericLiteLLMParams())
assert "Authorization" not in headers
def test_sigv4_is_skipped_when_a_bearer_token_is_present(self):
"""Bedrock API keys are Bearer; signing on top would be wrong."""
headers, body = _cfg().sign_request(
headers={"Authorization": "Bearer sk-bedrock"},
optional_params={},
request_data={},
api_base="https://bedrock-runtime.us-east-1.amazonaws.com/openai/v1/responses",
api_key="sk-bedrock",
)
assert headers["Authorization"] == "Bearer sk-bedrock"
assert body is None
class TestProviderIdentity:
def test_reports_the_bedrock_provider(self):
"""Cost tracking and callbacks key off this, so it must stay `bedrock` rather
than becoming a separate provider."""
assert _cfg().custom_llm_provider == LlmProviders.BEDROCK
class TestSigV4Fallback:
def test_signs_with_sigv4_when_no_bearer_token_is_present(self, monkeypatch):
"""No Bedrock API key means SigV4 over the standard credential chain. Static
credentials are set in the environment so signing stays a local computation."""
monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False)
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY")
monkeypatch.setenv("AWS_REGION_NAME", "us-east-1")
headers, body = _cfg().sign_request(
headers={"content-type": "application/json"},
optional_params={"aws_region_name": "us-east-1"},
request_data={"model": MODEL, "input": "hi"},
api_base="https://bedrock-runtime.us-east-1.amazonaws.com/openai/v1/responses",
api_key=None,
)
assert "Authorization" in headers
assert headers["Authorization"].startswith("AWS4-HMAC-SHA256")
assert "Credential=AKIAIOSFODNN7EXAMPLE" in headers["Authorization"]
class TestErrorClass:
"""Bedrock's request id must survive; the OpenAI base builds a blank response."""
def test_amzn_request_id_is_preserved(self):
error = _cfg().get_error_class(
error_message="boom",
status_code=500,
headers={"x-amzn-RequestId": "req-500"},
)
assert error.status_code == 500
assert error.response.headers["x-amzn-requestid"] == "req-500"
class TestPriceMapGate:
def test_absent_model_has_no_signal(self):
assert bedrock_supports_openai_responses(MODEL, {}) is False
def test_none_model_is_false(self):
assert bedrock_supports_openai_responses(None, {}) is False
def test_signal_on_the_bare_key(self):
cost = {MODEL: {"supported_endpoints": ["/v1/responses"]}}
assert bedrock_supports_openai_responses(MODEL, cost) is True
def test_signal_on_the_bedrock_prefixed_key(self):
cost = {f"bedrock/{MODEL}": {"supported_endpoints": ["/v1/responses"]}}
assert bedrock_supports_openai_responses(MODEL, cost) is True
def test_other_endpoints_do_not_count(self):
cost = {MODEL: {"supported_endpoints": ["/v1/messages"]}}
assert bedrock_supports_openai_responses(MODEL, cost) is False
class TestForModelGate:
"""The capability decision lives on the adapter, not in the shared dispatch."""
def test_returns_a_config_for_a_signalled_model(self):
with patch.object( # test-quality-ok: the gate reads the global cost map by design; no injection point exists
litellm, "model_cost", {MODEL: {"supported_endpoints": ["/v1/responses"]}}
):
assert isinstance(BedrockOpenAIResponsesConfig.for_model(MODEL), BedrockOpenAIResponsesConfig)
def test_returns_none_for_an_unsignalled_model(self):
with patch.object( # test-quality-ok: the gate reads the global cost map by design; no injection point exists
litellm, "model_cost", {}
):
assert BedrockOpenAIResponsesConfig.for_model(MODEL) is None
def test_returns_none_for_no_model(self):
with patch.object( # test-quality-ok: the gate reads the global cost map by design; no injection point exists
litellm, "model_cost", {}
):
assert BedrockOpenAIResponsesConfig.for_model(None) is None
class TestProviderResolution:
"""model_cost is patched explicitly: it is populated at import time from a GitHub
fetch unless LITELLM_LOCAL_MODEL_COST_MAP is set, and conftest's monkeypatch of
that variable lands after import — so these must not read the global."""
def test_signalled_model_resolves_to_the_bedrock_responses_config(self):
with patch.object( # test-quality-ok: resolution reads the global cost map by design; no HTTP boundary or injection point exists
litellm, "model_cost", {MODEL: {"supported_endpoints": ["/v1/responses"]}}
):
cfg = ProviderConfigManager.get_provider_responses_api_config(model=MODEL, provider=LlmProviders.BEDROCK)
assert isinstance(cfg, BedrockOpenAIResponsesConfig)
def test_unsignalled_model_keeps_the_existing_bridge(self):
"""Claude on Bedrock has no OpenAI surface; it must keep falling through to
the chat-completions bridge exactly as before."""
with patch.object(litellm, "model_cost", {}): # test-quality-ok: resolution reads the global cost map by design
cfg = ProviderConfigManager.get_provider_responses_api_config(
model="anthropic.claude-3-haiku-20240307-v1:0", provider=LlmProviders.BEDROCK
)
assert cfg is None
@pytest.mark.parametrize(
("family", "variants"),
[("gpt-5.6", ("sol", "terra", "luna")), ("gpt-6", ("astra", "sol", "luna"))],
)
def test_the_shipped_price_map_signals_the_openai_families(self, family: str, variants: tuple[str, ...]):
"""Reads the bundled backup directly rather than the network-fetched global."""
shipped = json.loads(
files("litellm").joinpath("model_prices_and_context_window_backup.json").read_text(encoding="utf-8")
)
for prefix in ("us", "global"):
for variant in variants:
model = f"{prefix}.openai.{family}-{variant}"
assert bedrock_supports_openai_responses(model, shipped) is True, model
class TestUnsupportedToolDrop:
"""Codex sends a web_search tool on every turn; bedrock-runtime 400s the whole request over it."""
_WEB_SEARCH_TOOL = {"type": "web_search", "external_web_access": False}
_SHELL_TOOL = {"type": "function", "name": "shell", "parameters": {"type": "object", "properties": {}}}
_NAMESPACE_TOOL = {
"type": "namespace",
"name": "multi_agent_v1",
"tools": [{"type": "function", "name": "spawn_agent"}],
}
def _outbound_tools(self, tools: list[dict]) -> object:
params = _cfg().map_openai_params(response_api_optional_params={"tools": tools}, model=MODEL, drop_params=False)
body = _cfg().transform_responses_api_request(
model=MODEL,
input="count the lines",
response_api_optional_request_params=params,
litellm_params=GenericLiteLLMParams(),
headers={},
)
return body.get("tools")
def test_codex_default_tools_reach_the_endpoint_without_web_search(self, caplog):
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
outbound = self._outbound_tools([self._SHELL_TOOL, self._WEB_SEARCH_TOOL, self._NAMESPACE_TOOL])
assert outbound == [self._SHELL_TOOL, self._NAMESPACE_TOOL]
dropped = [r.getMessage() for r in caplog.records if "dropping unsupported tool type" in r.getMessage()]
assert len(dropped) == 1 and "web_search" in dropped[0]
def test_only_unsupported_tools_means_no_tools_key(self):
assert self._outbound_tools([self._WEB_SEARCH_TOOL, {"type": "web_search_preview"}]) is None
def test_supported_tools_are_not_logged_as_dropped(self, caplog):
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
outbound = self._outbound_tools([self._SHELL_TOOL, {"type": "custom", "name": "exec"}])
assert outbound == [self._SHELL_TOOL, {"type": "custom", "name": "exec"}]
assert not [r for r in caplog.records if "dropping unsupported tool type" in r.getMessage()]
class TestFileSearchEmulation:
"""bedrock-runtime runs no server-side tools, so a file_search tool must take the emulated path."""
def test_file_search_tool_is_routed_to_emulation(self):
tools = [{"type": "file_search", "vector_store_ids": ["vs_1"]}]
assert should_use_emulated_file_search(tools, _cfg()) is True
def test_plain_function_tools_skip_emulation(self):
tools = [{"type": "function", "name": "shell", "parameters": {"type": "object", "properties": {}}}]
assert should_use_emulated_file_search(tools, _cfg()) is False
class TestCodexHistoryNormalization:
def test_history_items_the_endpoint_rejects_are_rewritten(self):
body = _cfg().transform_responses_api_request(
model=MODEL,
input=[
{"type": "agent_message", "content": [{"type": "output_text", "text": "prior"}]},
{"type": "context_compaction", "encrypted_content": "abc"},
{"type": "local_shell_call", "call_id": "c1", "action": {"command": ["ls"]}},
{"role": "user", "content": "carry on"},
],
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert [i.get("type") or i.get("role") for i in body["input"]] == [
"message",
"compaction",
"function_call",
"user",
]
def test_a_first_turn_request_is_untouched(self):
"""The rejected types are history items, so turn one exercises none of this."""
original = [{"role": "user", "content": "first turn"}]
body = _cfg().transform_responses_api_request(
model=MODEL,
input=list(original),
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert body["input"] == original
class TestBackgroundDrop:
"""The Converse bridge answered `background` requests synchronously; bedrock-runtime 400s the parameter."""
def test_background_is_dropped_with_a_warning(self, caplog):
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
params = _cfg().map_openai_params(
response_api_optional_params={"background": True, "max_output_tokens": 64},
model=MODEL,
drop_params=False,
)
assert params == {"max_output_tokens": 64}
dropped = [r.getMessage() for r in caplog.records if "dropping unsupported parameter" in r.getMessage()]
assert len(dropped) == 1 and "background" in dropped[0]
def test_without_background_nothing_is_dropped_or_logged(self, caplog):
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
params = _cfg().map_openai_params(
response_api_optional_params={"max_output_tokens": 64}, model=MODEL, drop_params=False
)
assert params == {"max_output_tokens": 64}
assert not [r for r in caplog.records if "dropping unsupported parameter" in r.getMessage()]
def _never_fetch(url: str) -> str:
raise AssertionError(f"unexpected sync fetch of {url}")
async def _never_fetch_async(url: str) -> str:
raise AssertionError(f"unexpected async fetch of {url}")
class TestRemoteImageInlining:
"""The Converse bridge downloaded http(s) image URLs; bedrock-runtime accepts only data: and s3://."""
_REMOTE = "https://example.com/grapes.png"
_DATA_URI = "data:image/png;base64,QUJD"
_INLINED = "data:image/png;base64,ZmV0Y2hlZA=="
def _input(self, remote: str) -> list[dict]:
return [
{
"role": "user",
"content": [
{"type": "input_text", "text": "What is this?"},
{"type": "input_image", "image_url": remote, "detail": "auto"},
{"type": "input_image", "image_url": remote},
{"type": "input_image", "image_url": self._DATA_URI},
{"type": "input_image", "image_url": "s3://bucket/grapes.png"},
{"type": "input_image", "file_id": "file-1"},
],
},
{"role": "assistant", "content": "plain string content"},
]
def test_sync_transform_fetches_each_remote_url_once_and_inlines_it(self):
fetched: list[str] = []
def fetch(url: str) -> str:
fetched.append(url)
return self._INLINED
body = BedrockOpenAIResponsesConfig(
fetch_image=fetch, async_fetch_image=_never_fetch_async
).transform_responses_api_request(
model=MODEL,
input=self._input(self._REMOTE),
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert body["input"] == self._input(self._INLINED)
assert fetched == [self._REMOTE]
def test_tool_output_lists_are_inlined_and_string_outputs_are_untouched(self):
fetched: list[str] = []
def fetch(url: str) -> str:
fetched.append(url)
return self._INLINED
def tool_turn(remote: str) -> list[dict]:
return [
{"type": "function_call", "call_id": "call_1", "name": "fetch_chart", "arguments": "{}"},
{
"type": "function_call_output",
"call_id": "call_1",
"output": [
{"type": "input_text", "text": "the chart"},
{"type": "input_image", "image_url": remote},
],
},
{"type": "function_call_output", "call_id": "call_2", "output": "https://example.com/plain-text.png"},
{"role": "user", "content": [{"type": "input_image", "image_url": remote}]},
]
body = BedrockOpenAIResponsesConfig(
fetch_image=fetch, async_fetch_image=_never_fetch_async
).transform_responses_api_request(
model=MODEL,
input=tool_turn(self._REMOTE),
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert body["input"] == tool_turn(self._INLINED)
assert fetched == [self._REMOTE]
def test_computer_screenshot_outputs_are_inlined(self):
fetched: list[str] = []
def fetch(url: str) -> str:
fetched.append(url)
return self._INLINED
def computer_turn(remote: str) -> list[dict]:
return [
{"type": "computer_call", "call_id": "call_1", "id": "cu_1", "actions": [{"type": "screenshot"}]},
{
"type": "computer_call_output",
"call_id": "call_1",
"output": {"type": "computer_screenshot", "image_url": remote},
},
{
"type": "computer_call_output",
"call_id": "call_2",
"output": {"type": "computer_screenshot", "file_id": "file-1"},
},
{
"type": "computer_call_output",
"call_id": "call_3",
"output": {"type": "computer_screenshot", "image_url": self._DATA_URI},
},
]
body = BedrockOpenAIResponsesConfig(
fetch_image=fetch, async_fetch_image=_never_fetch_async
).transform_responses_api_request(
model=MODEL,
input=computer_turn(self._REMOTE),
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert body["input"] == computer_turn(self._INLINED)
assert fetched == [self._REMOTE]
@pytest.mark.asyncio
async def test_async_transform_fetches_with_the_async_fetcher(self):
fetched: list[str] = []
async def fetch(url: str) -> str:
fetched.append(url)
return self._INLINED
body = await BedrockOpenAIResponsesConfig(
fetch_image=_never_fetch, async_fetch_image=fetch
).async_transform_responses_api_request(
model=MODEL,
input=self._input(self._REMOTE),
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert body["input"] == self._input(self._INLINED)
assert fetched == [self._REMOTE]
@pytest.mark.asyncio
async def test_inputs_without_remote_images_never_fetch(self):
cfg = BedrockOpenAIResponsesConfig(fetch_image=_never_fetch, async_fetch_image=_never_fetch_async)
local_only = self._input(self._DATA_URI)
sync_body = cfg.transform_responses_api_request(
model=MODEL,
input=local_only,
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
async_body = await cfg.async_transform_responses_api_request(
model=MODEL,
input="a plain string prompt",
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert sync_body["input"] == local_only
assert async_body["input"] == "a plain string prompt"
@pytest.mark.asyncio
async def test_inlining_runs_before_codex_history_normalization(self):
async def fetch(url: str) -> str:
return self._INLINED
body = await BedrockOpenAIResponsesConfig(
fetch_image=_never_fetch, async_fetch_image=fetch
).async_transform_responses_api_request(
model=MODEL,
input=[
{"type": "agent_message", "content": [{"type": "output_text", "text": "prior"}]},
{"role": "user", "content": [{"type": "input_image", "image_url": self._REMOTE}]},
],
response_api_optional_request_params={},
litellm_params=GenericLiteLLMParams(),
headers={},
)
assert [i.get("type") or i.get("role") for i in body["input"]] == ["message", "user"]
assert body["input"][1]["content"] == [{"type": "input_image", "image_url": self._INLINED}]

47
tests/test_litellm/llms/custom_httpx/test_llm_http_handler.py
View file

@ -407,11 +407,9 @@ async def test_async_response_api_handler_streams_when_provider_transform_adds_s
config = Mock()
config.validate_environment.return_value = {}
config.get_complete_url.return_value = "https://chatgpt.example.com/responses"
config.transform_responses_api_request.return_value = {
"model": "gpt-5.3-codex",
"input": "hi",
"stream": True,
}
config.async_transform_responses_api_request = AsyncMock(
return_value={"model": "gpt-5.3-codex", "input": "hi", "stream": True}
)
config.sign_request.return_value = ({}, None)
client = AsyncHTTPHandler()
client.post = AsyncMock(
@ -447,7 +445,9 @@ async def test_async_response_api_handler_streaming_passes_logging_obj_to_post()
config = Mock()
config.validate_environment.return_value = {}
config.get_complete_url.return_value = "https://chatgpt.example.com/responses"
config.transform_responses_api_request.return_value = {"model": "gpt-5", "input": "hi", "stream": True}
config.async_transform_responses_api_request = AsyncMock(
return_value={"model": "gpt-5", "input": "hi", "stream": True}
)
config.sign_request.return_value = ({}, None)
client = AsyncHTTPHandler()
client.post = AsyncMock(
@ -472,6 +472,41 @@ async def test_async_response_api_handler_streaming_passes_logging_obj_to_post()
assert client.post.call_args.kwargs["logging_obj"] is logging_obj
@pytest.mark.asyncio
async def test_async_response_api_handler_posts_the_async_transform_hook_result():
"""A provider whose request transform must await (Bedrock inlines remote image URLs)
overrides the async hook; the async handler has to send that result, not the sync one."""
handler = BaseLLMHTTPHandler()
config = Mock()
config.validate_environment.return_value = {}
config.get_complete_url.return_value = "https://chatgpt.example.com/responses"
config.async_transform_responses_api_request = AsyncMock(
return_value={"model": "gpt-5", "input": "inlined by the async hook", "stream": True}
)
config.sign_request.return_value = ({}, None)
client = AsyncHTTPHandler()
client.post = AsyncMock(
return_value=httpx.Response(
200,
request=httpx.Request("POST", "https://chatgpt.example.com/responses"),
)
)
await handler.async_response_api_handler(
model="gpt-5",
input="hi",
responses_api_provider_config=config,
response_api_optional_request_params={},
custom_llm_provider="chatgpt",
litellm_params=GenericLiteLLMParams(),
logging_obj=Mock(),
client=client,
)
assert client.post.call_args.kwargs["json"]["input"] == "inlined by the async hook"
config.transform_responses_api_request.assert_not_called()
@pytest.mark.asyncio
async def test_async_responses_records_llm_api_duration():
"""aresponses must feed the httpx timing into the logging obj, so the proxy can emit