feat(bedrock_mantle): add SigV4 sign_request via composed BaseAWSLLM (bearer path)

litellm/llms/bedrock_mantle/responses/transformation.py

@@ -10,8 +10,12 @@ Auth: AWS Bedrock API key as Bearer token (BEDROCK_MANTLE_API_KEY or the
 standard AWS_BEARER_TOKEN_BEDROCK), NOT SigV4.
 """
 
-from typing import Optional
+import re
+from typing import Optional, Tuple
 
+from botocore.exceptions import NoCredentialsError
+
+from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
 from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
 from litellm.secret_managers.main import get_secret_str
 from litellm.types.router import GenericLiteLLMParams
@@ -29,22 +33,44 @@ _BASE_SUFFIXES_TO_STRIP = (
     "/v1",
 )
 
+# Standard Mantle host: https://bedrock-mantle.<region>.api.aws (group 1 = region).
+_MANTLE_HOST_RE = re.compile(
+    r"^https?://bedrock-mantle\.([^/.]+)\.api\.aws", re.IGNORECASE
+)
+
 
 class BedrockMantleResponsesAPIConfig(OpenAIResponsesAPIConfig):
+    def __init__(self, aws_signer: Optional[BaseAWSLLM] = None):
+        super().__init__()
+        self._aws_signer = aws_signer or BaseAWSLLM()
+
     @property
     def custom_llm_provider(self) -> LlmProviders:
         return LlmProviders.BEDROCK_MANTLE
 
+    @staticmethod
+    def _resolve_region(params: dict) -> str:
+        region = params.get("aws_region_name")
+        if region:
+            return region
+        base = params.get("api_base") or get_secret_str("BEDROCK_MANTLE_API_BASE")
+        if base:
+            match = _MANTLE_HOST_RE.match(base.rstrip("/"))
+            if match:
+                return match.group(1)
+        return (
+            get_secret_str("BEDROCK_MANTLE_REGION")
+            or get_secret_str("AWS_REGION_NAME")
+            or get_secret_str("AWS_REGION")
+            or BEDROCK_MANTLE_DEFAULT_REGION
+        )
+
     def get_complete_url(
         self,
         api_base: Optional[str],
         litellm_params: dict,
     ) -> str:
-        region = (
-            get_secret_str("BEDROCK_MANTLE_REGION")
-            or get_secret_str("AWS_REGION")
-            or BEDROCK_MANTLE_DEFAULT_REGION
-        )
+        region = self._resolve_region({**litellm_params, "api_base": api_base})
         base = (
             api_base
             or get_secret_str("BEDROCK_MANTLE_API_BASE")
@@ -55,6 +81,11 @@ class BedrockMantleResponsesAPIConfig(OpenAIResponsesAPIConfig):
             if base.endswith(suffix):
                 base = base[: -len(suffix)]
                 break
+        # For the standard Mantle host (including the default-region base that
+        # responses/main.py auto-injects into litellm_params.api_base), pin to the
+        # single resolved region so aws_region_name wins; preserve custom proxy hosts.
+        if _MANTLE_HOST_RE.match(base):
+            base = f"https://bedrock-mantle.{region}.api.aws"
         return f"{base}/openai/v1/responses"
 
     def validate_environment(
@@ -79,3 +110,53 @@ class BedrockMantleResponsesAPIConfig(OpenAIResponsesAPIConfig):
 
     def supports_native_websocket(self) -> bool:
         return False
+
+    def sign_request(
+        self,
+        headers: dict,
+        optional_params: dict,
+        request_data: dict,
+        api_base: str,
+        api_key: Optional[str] = None,
+        model: Optional[str] = None,
+        stream: Optional[bool] = None,
+        fake_stream: Optional[bool] = None,
+    ) -> Tuple[dict, Optional[bytes]]:
+        bearer = (
+            api_key
+            or get_secret_str("BEDROCK_MANTLE_API_KEY")
+            or get_secret_str("AWS_BEARER_TOKEN_BEDROCK")
+        )
+        if not bearer:
+            # SigV4 path. Pin the credential-scope region to the region of the actual
+            # signing URL (api_base, already region-resolved by get_complete_url) so the
+            # SigV4 scope and the URL host can never disagree. Resolve from api_base first,
+            # then fall back to the regular precedence. Also drop any caller Authorization
+            # so _sign_request's restore-original-Authorization step cannot override the
+            # SigV4 header.
+            optional_params = {
+                **optional_params,
+                "aws_region_name": self._resolve_region(
+                    {**optional_params, "api_base": api_base}
+                ),
+            }
+            headers = {k: v for k, v in headers.items() if k.lower() != "authorization"}
+        try:
+            return self._aws_signer._sign_request(
+                service_name="bedrock",
+                headers=headers,
+                optional_params=optional_params,
+                request_data=request_data,
+                api_base=api_base,
+                api_key=bearer,
+                model=model,
+                stream=stream,
+                fake_stream=fake_stream,
+            )
+        except NoCredentialsError as e:
+            raise ValueError(
+                "Bedrock Mantle auth failed: no Bearer token and no usable AWS "
+                "credentials. Set BEDROCK_MANTLE_API_KEY (or AWS_BEARER_TOKEN_BEDROCK) "
+                "or pass api_key for Bearer auth, or provide AWS credentials "
+                "(IAM role / access key / profile / web identity) for SigV4."
+            ) from e

tests/test_litellm/llms/bedrock_mantle/test_bedrock_mantle_responses_transformation.py

@@ -261,6 +261,54 @@ def local_cost_map(monkeypatch):
         litellm.get_model_info.cache_clear()
 
 
+class TestBedrockMantleResponsesSigV4:
+    def test_bearer_short_circuits_without_credentials(self, monkeypatch):
+        from unittest.mock import MagicMock
+        from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
+
+        monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False)
+        monkeypatch.delenv("BEDROCK_MANTLE_API_KEY", raising=False)
+
+        signer = BaseAWSLLM()
+        signer.get_credentials = MagicMock(
+            side_effect=AssertionError("get_credentials must not run for bearer auth")
+        )
+        cfg = BedrockMantleResponsesAPIConfig(aws_signer=signer)
+
+        headers, signed_body = cfg.sign_request(
+            headers={},
+            optional_params={},
+            request_data={"input": "hi"},
+            api_base="https://bedrock-mantle.us-east-2.api.aws/openai/v1/responses",
+            api_key="bearer-from-config",
+        )
+        assert headers["Authorization"] == "Bearer bearer-from-config"
+        assert signed_body == b'{"input": "hi"}'
+        signer.get_credentials.assert_not_called()
+
+    def test_bearer_resolved_from_mantle_env_key(self, monkeypatch):
+        from unittest.mock import MagicMock
+        from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
+
+        monkeypatch.delenv("AWS_BEARER_TOKEN_BEDROCK", raising=False)
+        monkeypatch.setenv("BEDROCK_MANTLE_API_KEY", "env-bearer")
+
+        signer = BaseAWSLLM()
+        signer.get_credentials = MagicMock(
+            side_effect=AssertionError("get_credentials must not run for bearer auth")
+        )
+        cfg = BedrockMantleResponsesAPIConfig(aws_signer=signer)
+
+        headers, _ = cfg.sign_request(
+            headers={},
+            optional_params={},
+            request_data={"input": "hi"},
+            api_base="https://bedrock-mantle.us-east-2.api.aws/openai/v1/responses",
+            api_key=None,
+        )
+        assert headers["Authorization"] == "Bearer env-bearer"
+
+
 class TestBedrockMantleResponsesPricing:
     def test_gpt_5_5_pricing_and_mode(self, local_cost_map):
         info = litellm.get_model_info("bedrock_mantle/openai.gpt-5.5")