fix(bedrock): treat an empty api_base or runtime endpoint as unset

litellm/llms/bedrock/base_aws_llm.py

@@ -1535,28 +1535,13 @@ class BaseAWSLLM(SignsRequestsWithAWS):
         aws_region_name: str,
         endpoint_type: Literal["runtime", "agent", "agentcore"] | None = "runtime",
     ) -> tuple[str, str]:
-        env_aws_bedrock_runtime_endpoint: Final = get_secret("AWS_BEDROCK_RUNTIME_ENDPOINT")
+        runtime_endpoint: Final = aws_bedrock_runtime_endpoint or get_secret_str("AWS_BEDROCK_RUNTIME_ENDPOINT")
-        if api_base is not None:
+        endpoint_url: Final = (
-            endpoint_url = api_base
+            api_base
-        elif aws_bedrock_runtime_endpoint is not None and isinstance(aws_bedrock_runtime_endpoint, str):
+            or runtime_endpoint
-            endpoint_url = aws_bedrock_runtime_endpoint
+            or self._select_default_endpoint_url(endpoint_type=endpoint_type, aws_region_name=aws_region_name)
-        elif env_aws_bedrock_runtime_endpoint and isinstance(env_aws_bedrock_runtime_endpoint, str):
+        )
-            endpoint_url = env_aws_bedrock_runtime_endpoint
+        return endpoint_url, runtime_endpoint or endpoint_url
-        else:
-            endpoint_url = self._select_default_endpoint_url(
-                endpoint_type=endpoint_type,
-                aws_region_name=aws_region_name,
-            )
-
-        # Determine proxy_endpoint_url
-        if aws_bedrock_runtime_endpoint is not None and isinstance(aws_bedrock_runtime_endpoint, str):
-            proxy_endpoint_url = aws_bedrock_runtime_endpoint
-        elif env_aws_bedrock_runtime_endpoint and isinstance(env_aws_bedrock_runtime_endpoint, str):
-            proxy_endpoint_url = env_aws_bedrock_runtime_endpoint
-        else:
-            proxy_endpoint_url = endpoint_url
-
-        return endpoint_url, proxy_endpoint_url
 
     def _select_default_endpoint_url(
         self,

tests/unit/llms/bedrock/count_tokens/test_bedrock_token_counter.py

@@ -91,6 +91,36 @@ async def test_bedrock_runtime_count_is_kept_when_it_answers() -> None:
     assert bedrock.posted_hosts() == (RUNTIME_HOST,)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("model_to_use", "runtime", "expected_hosts", "expected_tokens"),
+    (
+        ("us.anthropic.claude-sonnet-4-6", (200, {"inputTokens": 1353}), (RUNTIME_HOST,), 1353),
+        ("us.anthropic.claude-opus-5-5", (400, UNSUPPORTED), (RUNTIME_HOST, MANTLE_HOST), 2177),
+    ),
+)
+async def test_deployment_with_an_empty_api_base_counts_on_the_default_runtime_host(
+    model_to_use: str,
+    runtime: tuple[int, Mapping[str, object]],
+    expected_hosts: tuple[str, ...],
+    expected_tokens: int,
+) -> None:
+    bedrock: Final = _Bedrock(runtime=runtime, mantle=(200, {"input_tokens": 2177}))
+
+    result: Final = await _counter(bedrock).count_tokens(
+        model_to_use=model_to_use,
+        messages=MESSAGES,
+        contents=None,
+        deployment={"litellm_params": {**DEPLOYMENT["litellm_params"], "api_base": ""}},
+        request_model=model_to_use,
+    )
+
+    assert result is not None
+    assert result.error is False
+    assert result.total_tokens == expected_tokens
+    assert bedrock.posted_hosts() == expected_hosts
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("model_to_use", "runtime"),

tests/unit/llms/bedrock/test_base_aws_llm.py

@@ -3848,6 +3848,39 @@ def test_get_runtime_endpoint(base_aws_llm):
     assert endpoint_url == "https://bedrock-runtime.us-east-1.amazonaws.com"
     assert proxy_endpoint_url == "https://bedrock-runtime.us-east-1.amazonaws.com"
 
+@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
+@pytest.mark.parametrize(
+    ("api_base", "aws_bedrock_runtime_endpoint", "env_runtime_endpoint", "expected"),
+    (
+        ("", None, None, "https://bedrock-runtime.us-east-1.amazonaws.com"),
+        (None, "", None, "https://bedrock-runtime.us-east-1.amazonaws.com"),
+        ("", "", None, "https://bedrock-runtime.us-east-1.amazonaws.com"),
+        ("", "https://vpce-runtime.example.com", None, "https://vpce-runtime.example.com"),
+        ("", None, "https://env-runtime.example.com", "https://env-runtime.example.com"),
+        (None, "", "https://env-runtime.example.com", "https://env-runtime.example.com"),
+    ),
+)
+def test_get_runtime_endpoint_treats_an_empty_setting_as_unset(
+    base_aws_llm: BaseAWSLLM,
+    monkeypatch: pytest.MonkeyPatch,
+    api_base: str | None,
+    aws_bedrock_runtime_endpoint: str | None,
+    env_runtime_endpoint: str | None,
+    expected: str,
+) -> None:
+    if env_runtime_endpoint is None:
+        monkeypatch.delenv("AWS_BEDROCK_RUNTIME_ENDPOINT", raising=False)
+    else:
+        monkeypatch.setenv("AWS_BEDROCK_RUNTIME_ENDPOINT", env_runtime_endpoint)
+
+    endpoints: Final = base_aws_llm.get_runtime_endpoint(
+        api_base=api_base,
+        aws_bedrock_runtime_endpoint=aws_bedrock_runtime_endpoint,
+        aws_region_name="us-east-1",
+    )
+
+    assert endpoints == (expected, expected)
+
 @pytest.fixture
 def clear_cache(base_aws_llm):
     """Clear the cache before each test"""