feat(lighthouse): Support Amazon Bedrock Long-Term API Key (#9343)

Co-authored-by: Chandrapal Badshah <12944530+Chan9390@users.noreply.github.com>

api/src/backend/tasks/jobs/lighthouse_providers.py

@@ -2,6 +2,8 @@ from typing import Dict
 
 import boto3
 import openai
+from botocore import UNSIGNED
+from botocore.config import Config
 from botocore.exceptions import BotoCoreError, ClientError
 from celery.utils.log import get_task_logger
 
@@ -56,21 +58,39 @@ def _extract_bedrock_credentials(
     """
     Safely extract AWS Bedrock credentials from a provider configuration.
 
+    Supports two authentication methods:
+    1. AWS access key + secret key + region
+    2. Bedrock API key (bearer token) + region
+
     Args:
         provider_cfg (LighthouseProviderConfiguration): The provider configuration instance
             containing the credentials.
 
     Returns:
-        Dict[str, str] | None: Dictionary with 'access_key_id', 'secret_access_key', and
-            'region' if present and valid, otherwise None.
+        Dict[str, str] | None: Dictionary with either:
+            - 'access_key_id', 'secret_access_key', and 'region' for access key auth
+            - 'api_key' and 'region' for API key (bearer token) auth
+            Returns None if credentials are invalid or missing.
     """
     creds = provider_cfg.credentials_decoded
     if not isinstance(creds, dict):
         return None
 
+    region = creds.get("region")
+    if not isinstance(region, str) or not region:
+        return None
+
+    # Check for API key authentication first
+    api_key = creds.get("api_key")
+    if isinstance(api_key, str) and api_key:
+        return {
+            "api_key": api_key,
+            "region": region,
+        }
+
+    # Fall back to access key authentication
     access_key_id = creds.get("access_key_id")
     secret_access_key = creds.get("secret_access_key")
-    region = creds.get("region")
 
     # Validate all required fields are present and are strings
     if (
@@ -78,8 +98,6 @@ def _extract_bedrock_credentials(
         or not access_key_id
         or not isinstance(secret_access_key, str)
         or not secret_access_key
-        or not isinstance(region, str)
-        or not region
     ):
         return None
 
@@ -90,6 +108,51 @@ def _extract_bedrock_credentials(
     }
 
 
+def _create_bedrock_client(
+    bedrock_creds: Dict[str, str], service_name: str = "bedrock"
+):
+    """
+    Create a boto3 Bedrock client with the appropriate authentication method.
+
+    Supports two authentication methods:
+    1. API key (bearer token) - uses unsigned requests with Authorization header
+    2. AWS access key + secret key - uses standard SigV4 signing
+
+    Args:
+        bedrock_creds: Dictionary with either:
+            - 'api_key' and 'region' for API key (bearer token) auth
+            - 'access_key_id', 'secret_access_key', and 'region' for access key auth
+        service_name: The Bedrock service name. Use 'bedrock' for control plane
+            operations (list_foundation_models, etc.) or 'bedrock-runtime' for
+            inference operations.
+
+    Returns:
+        boto3 client configured for the specified Bedrock service.
+    """
+    region = bedrock_creds["region"]
+
+    if "api_key" in bedrock_creds:
+        bearer_token = bedrock_creds["api_key"]
+        client = boto3.client(
+            service_name=service_name,
+            region_name=region,
+            config=Config(signature_version=UNSIGNED),
+        )
+
+        def inject_bearer_token(request, **kwargs):
+            request.headers["Authorization"] = f"Bearer {bearer_token}"
+
+        client.meta.events.register("before-send.*.*", inject_bearer_token)
+        return client
+
+    return boto3.client(
+        service_name=service_name,
+        region_name=region,
+        aws_access_key_id=bedrock_creds["access_key_id"],
+        aws_secret_access_key=bedrock_creds["secret_access_key"],
+    )
+
+
 def check_lighthouse_provider_connection(provider_config_id: str) -> Dict:
     """
     Validate a Lighthouse provider configuration by calling the provider API and
@@ -141,12 +204,7 @@ def check_lighthouse_provider_connection(provider_config_id: str) -> Dict:
                 }
 
             # Test connection by listing foundation models
-            bedrock_client = boto3.client(
-                "bedrock",
-                aws_access_key_id=bedrock_creds["access_key_id"],
-                aws_secret_access_key=bedrock_creds["secret_access_key"],
-                region_name=bedrock_creds["region"],
-            )
+            bedrock_client = _create_bedrock_client(bedrock_creds)
             _ = bedrock_client.list_foundation_models()
 
         elif (
@@ -232,105 +290,219 @@ def _fetch_openai_compatible_models(base_url: str, api_key: str) -> Dict[str, st
     return available_models
 
 
-def _fetch_bedrock_models(bedrock_creds: Dict[str, str]) -> Dict[str, str]:
+def _get_region_prefix(region: str) -> str:
     """
-    Fetch available models from AWS Bedrock with entitlement verification.
+    Determine geographic prefix for AWS region.
 
-    This function:
-    1. Lists foundation models with TEXT modality support
-    2. Lists inference profiles with TEXT modality support
-    3. Verifies user has entitlement access to each model
+    Examples: ap-south-1 -> apac, us-east-1 -> us, eu-west-1 -> eu
+    """
+    if region.startswith(("us-", "ca-", "sa-")):
+        return "us"
+    elif region.startswith("eu-"):
+        return "eu"
+    elif region.startswith("ap-"):
+        return "apac"
+    return "global"
 
-    Args:
-        bedrock_creds: Dictionary with 'access_key_id', 'secret_access_key', and 'region'.
+
+def _clean_inference_profile_name(profile_name: str) -> str:
+    """
+    Remove geographic prefix from inference profile name.
+
+    AWS includes geographic prefixes in profile names which are redundant
+    since the profile ID already contains this information.
+
+    Examples:
+        "APAC Anthropic Claude 3.5 Sonnet" -> "Anthropic Claude 3.5 Sonnet"
+        "GLOBAL Claude Sonnet 4.5" -> "Claude Sonnet 4.5"
+        "US Anthropic Claude 3 Haiku" -> "Anthropic Claude 3 Haiku"
+    """
+    prefixes = ["APAC ", "GLOBAL ", "US ", "EU ", "APAC-", "GLOBAL-", "US-", "EU-"]
+
+    for prefix in prefixes:
+        if profile_name.upper().startswith(prefix.upper()):
+            return profile_name[len(prefix) :].strip()
+
+    return profile_name
+
+
+def _supports_text_modality(input_modalities: list, output_modalities: list) -> bool:
+    """Check if model supports TEXT for both input and output."""
+    return "TEXT" in input_modalities and "TEXT" in output_modalities
+
+
+def _get_foundation_model_modalities(
+    bedrock_client, model_id: str
+) -> tuple[list, list] | None:
+    """
+    Fetch input and output modalities for a foundation model.
 
     Returns:
-        Dict mapping model_id to model_name for all accessible models.
-
-    Raises:
-        BotoCoreError, ClientError: If AWS API calls fail.
+        (input_modalities, output_modalities) or None if fetch fails
     """
-    bedrock_client = boto3.client(
-        "bedrock",
-        aws_access_key_id=bedrock_creds["access_key_id"],
-        aws_secret_access_key=bedrock_creds["secret_access_key"],
-        region_name=bedrock_creds["region"],
-    )
+    try:
+        model_info = bedrock_client.get_foundation_model(modelIdentifier=model_id)
+        model_details = model_info.get("modelDetails", {})
+        input_mods = model_details.get("inputModalities", [])
+        output_mods = model_details.get("outputModalities", [])
+        return (input_mods, output_mods)
+    except (BotoCoreError, ClientError) as e:
+        logger.debug("Could not fetch model details for %s: %s", model_id, str(e))
+        return None
 
-    models_to_check: Dict[str, str] = {}
 
-    # Step 1: Get foundation models with TEXT modality
+def _extract_foundation_model_ids(profile_models: list) -> list[str]:
+    """
+    Extract foundation model IDs from inference profile model ARNs.
+
+    Args:
+        profile_models: List of model references from inference profile
+
+    Returns:
+        List of foundation model IDs extracted from ARNs
+    """
+    model_ids = []
+    for model_ref in profile_models:
+        model_arn = model_ref.get("modelArn", "")
+        if "foundation-model/" in model_arn:
+            model_id = model_arn.split("foundation-model/")[1]
+            model_ids.append(model_id)
+    return model_ids
+
+
+def _build_inference_profile_map(
+    bedrock_client, region: str
+) -> Dict[str, tuple[str, str]]:
+    """
+    Build map of foundation_model_id -> best inference profile.
+
+    Returns:
+        Dict mapping foundation_model_id to (profile_id, profile_name)
+        Only includes profiles with TEXT modality support
+        Prefers region-matched profiles over others
+    """
+    region_prefix = _get_region_prefix(region)
+    model_to_profile: Dict[str, tuple[str, str]] = {}
+
+    try:
+        response = bedrock_client.list_inference_profiles()
+        profiles = response.get("inferenceProfileSummaries", [])
+
+        for profile in profiles:
+            profile_id = profile.get("inferenceProfileId")
+            profile_name = profile.get("inferenceProfileName")
+
+            if not profile_id or not profile_name:
+                continue
+
+            profile_models = profile.get("models", [])
+            if not profile_models:
+                continue
+
+            foundation_model_ids = _extract_foundation_model_ids(profile_models)
+            if not foundation_model_ids:
+                continue
+
+            modalities = _get_foundation_model_modalities(
+                bedrock_client, foundation_model_ids[0]
+            )
+            if not modalities:
+                continue
+
+            input_mods, output_mods = modalities
+            if not _supports_text_modality(input_mods, output_mods):
+                continue
+
+            is_preferred = profile_id.startswith(f"{region_prefix}.")
+            clean_name = _clean_inference_profile_name(profile_name)
+
+            for foundation_model_id in foundation_model_ids:
+                if foundation_model_id not in model_to_profile:
+                    model_to_profile[foundation_model_id] = (profile_id, clean_name)
+                elif is_preferred and not model_to_profile[foundation_model_id][
+                    0
+                ].startswith(f"{region_prefix}."):
+                    model_to_profile[foundation_model_id] = (profile_id, clean_name)
+
+    except (BotoCoreError, ClientError) as e:
+        logger.info("Could not fetch inference profiles in %s: %s", region, str(e))
+
+    return model_to_profile
+
+
+def _check_on_demand_availability(bedrock_client, model_id: str) -> bool:
+    """Check if an ON_DEMAND foundation model is entitled and available."""
+    try:
+        availability = bedrock_client.get_foundation_model_availability(
+            modelId=model_id
+        )
+        entitlement = availability.get("entitlementAvailability")
+        return entitlement == "AVAILABLE"
+    except (BotoCoreError, ClientError) as e:
+        logger.debug("Could not check availability for %s: %s", model_id, str(e))
+        return False
+
+
+def _fetch_bedrock_models(bedrock_creds: Dict[str, str]) -> Dict[str, str]:
+    """
+    Fetch available models from AWS Bedrock, preferring inference profiles over ON_DEMAND.
+
+    Strategy:
+    1. Build map of foundation_model -> best_inference_profile (with TEXT validation)
+    2. For each TEXT-capable foundation model:
+       - Use inference profile ID if available (preferred - better throughput)
+       - Fallback to foundation model ID if only ON_DEMAND available
+    3. Verify entitlement for ON_DEMAND models
+
+    Args:
+        bedrock_creds: Dict with 'region' and auth credentials
+
+    Returns:
+        Dict mapping model_id to model_name. IDs can be:
+        - Inference profile IDs (e.g., "apac.anthropic.claude-3-5-sonnet-20240620-v1:0")
+        - Foundation model IDs (e.g., "anthropic.claude-3-5-sonnet-20240620-v1:0")
+    """
+    bedrock_client = _create_bedrock_client(bedrock_creds)
+    region = bedrock_creds["region"]
+
+    model_to_profile = _build_inference_profile_map(bedrock_client, region)
+
     foundation_response = bedrock_client.list_foundation_models()
     model_summaries = foundation_response.get("modelSummaries", [])
 
-    for model in model_summaries:
-        # Check if model supports TEXT input and output modality
-        input_modalities = model.get("inputModalities", [])
-        output_modalities = model.get("outputModalities", [])
+    models_to_return: Dict[str, str] = {}
+    on_demand_models: set[str] = set()
 
-        if "TEXT" not in input_modalities or "TEXT" not in output_modalities:
+    for model in model_summaries:
+        input_mods = model.get("inputModalities", [])
+        output_mods = model.get("outputModalities", [])
+
+        if not _supports_text_modality(input_mods, output_mods):
             continue
 
         model_id = model.get("modelId")
-        if not model_id:
+        model_name = model.get("modelName")
+
+        if not model_id or not model_name:
             continue
 
-        inference_types = model.get("inferenceTypesSupported", [])
+        if model_id in model_to_profile:
+            profile_id, profile_name = model_to_profile[model_id]
+            models_to_return[profile_id] = profile_name
+        else:
+            inference_types = model.get("inferenceTypesSupported", [])
+            if "ON_DEMAND" in inference_types:
+                models_to_return[model_id] = model_name
+                on_demand_models.add(model_id)
 
-        # Only include models with ON_DEMAND inference support
-        if "ON_DEMAND" in inference_types:
-            models_to_check[model_id] = model["modelName"]
-
-    # Step 2: Get inference profiles
-    try:
-        inference_profiles_response = bedrock_client.list_inference_profiles()
-        inference_profiles = inference_profiles_response.get(
-            "inferenceProfileSummaries", []
-        )
-
-        for profile in inference_profiles:
-            # Check if profile supports TEXT modality
-            input_modalities = profile.get("inputModalities", [])
-            output_modalities = profile.get("outputModalities", [])
-
-            if "TEXT" not in input_modalities or "TEXT" not in output_modalities:
-                continue
-
-            profile_id = profile.get("inferenceProfileId")
-            if profile_id:
-                models_to_check[profile_id] = profile["inferenceProfileName"]
-
-    except (BotoCoreError, ClientError) as e:
-        logger.info(
-            "Could not fetch inference profiles in %s: %s",
-            bedrock_creds["region"],
-            str(e),
-        )
-
-    # Step 3: Verify entitlement availability for each model
     available_models: Dict[str, str] = {}
 
-    for model_id, model_name in models_to_check.items():
-        try:
-            availability = bedrock_client.get_foundation_model_availability(
-                modelId=model_id
-            )
-
-            entitlement = availability.get("entitlementAvailability")
-
-            # Only include models user has access to
-            if entitlement == "AVAILABLE":
+    for model_id, model_name in models_to_return.items():
+        if model_id in on_demand_models:
+            if _check_on_demand_availability(bedrock_client, model_id):
                 available_models[model_id] = model_name
-            else:
-                logger.debug(
-                    "Skipping model %s - entitlement status: %s", model_id, entitlement
-                )
-
-        except (BotoCoreError, ClientError) as e:
-            logger.debug(
-                "Could not check availability for model %s: %s", model_id, str(e)
-            )
-            continue
+        else:
+            available_models[model_id] = model_name
 
     return available_models
 
@@ -359,7 +531,6 @@ def refresh_lighthouse_provider_models(provider_config_id: str) -> Dict:
     provider_cfg = LighthouseProviderConfiguration.objects.get(pk=provider_config_id)
     fetched_models: Dict[str, str] = {}
 
-    # Fetch models from the appropriate provider
     try:
         if (
             provider_cfg.provider_type

api/src/backend/tasks/tests/test_tasks.py

@@ -4,6 +4,10 @@ from unittest.mock import MagicMock, patch
 import openai
 import pytest
 from botocore.exceptions import ClientError
+from tasks.jobs.lighthouse_providers import (
+    _create_bedrock_client,
+    _extract_bedrock_credentials,
+)
 from tasks.tasks import (
     _perform_scan_complete_tasks,
     check_integrations_task,
@@ -21,6 +25,198 @@ from api.models import (
 )
 
 
+@pytest.mark.django_db
+class TestExtractBedrockCredentials:
+    """Unit tests for _extract_bedrock_credentials helper function."""
+
+    def test_extract_access_key_credentials(self, tenants_fixture):
+        """Test extraction of access key + secret key credentials."""
+        provider_cfg = LighthouseProviderConfiguration(
+            tenant_id=tenants_fixture[0].id,
+            provider_type=LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+            is_active=True,
+        )
+        provider_cfg.credentials_decoded = {
+            "access_key_id": "AKIAIOSFODNN7EXAMPLE",
+            "secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+            "region": "us-east-1",
+        }
+        provider_cfg.save()
+
+        result = _extract_bedrock_credentials(provider_cfg)
+
+        assert result is not None
+        assert result["access_key_id"] == "AKIAIOSFODNN7EXAMPLE"
+        assert result["secret_access_key"] == "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+        assert result["region"] == "us-east-1"
+        assert "api_key" not in result
+
+    def test_extract_api_key_credentials(self, tenants_fixture):
+        """Test extraction of API key (bearer token) credentials."""
+        valid_api_key = "ABSKQmVkcm9ja0FQSUtleS" + ("A" * 110)
+        provider_cfg = LighthouseProviderConfiguration(
+            tenant_id=tenants_fixture[0].id,
+            provider_type=LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+            is_active=True,
+        )
+        provider_cfg.credentials_decoded = {
+            "api_key": valid_api_key,
+            "region": "us-west-2",
+        }
+        provider_cfg.save()
+
+        result = _extract_bedrock_credentials(provider_cfg)
+
+        assert result is not None
+        assert result["api_key"] == valid_api_key
+        assert result["region"] == "us-west-2"
+        assert "access_key_id" not in result
+        assert "secret_access_key" not in result
+
+    def test_api_key_takes_precedence_over_access_keys(self, tenants_fixture):
+        """Test that API key is preferred when both auth methods are present."""
+        valid_api_key = "ABSKQmVkcm9ja0FQSUtleS" + ("B" * 110)
+        provider_cfg = LighthouseProviderConfiguration(
+            tenant_id=tenants_fixture[0].id,
+            provider_type=LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+            is_active=True,
+        )
+        provider_cfg.credentials_decoded = {
+            "api_key": valid_api_key,
+            "access_key_id": "AKIAIOSFODNN7EXAMPLE",
+            "secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+            "region": "eu-west-1",
+        }
+        provider_cfg.save()
+
+        result = _extract_bedrock_credentials(provider_cfg)
+
+        assert result is not None
+        assert result["api_key"] == valid_api_key
+        assert result["region"] == "eu-west-1"
+        assert "access_key_id" not in result
+
+    def test_missing_region_returns_none(self, tenants_fixture):
+        """Test that missing region returns None."""
+        provider_cfg = LighthouseProviderConfiguration(
+            tenant_id=tenants_fixture[0].id,
+            provider_type=LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+            is_active=True,
+        )
+        provider_cfg.credentials_decoded = {
+            "api_key": "ABSKQmVkcm9ja0FQSUtleS" + ("A" * 110),
+        }
+        provider_cfg.save()
+
+        result = _extract_bedrock_credentials(provider_cfg)
+
+        assert result is None
+
+    def test_empty_credentials_returns_none(self, tenants_fixture):
+        """Test that empty credentials dict returns None (region only is not enough)."""
+        provider_cfg = LighthouseProviderConfiguration(
+            tenant_id=tenants_fixture[0].id,
+            provider_type=LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+            is_active=True,
+        )
+        # Only region, no auth credentials - should return None
+        provider_cfg.credentials_decoded = {
+            "region": "us-east-1",
+        }
+        provider_cfg.save()
+
+        result = _extract_bedrock_credentials(provider_cfg)
+
+        assert result is None
+
+    def test_non_dict_credentials_returns_none(self, tenants_fixture):
+        """Test that non-dict credentials returns None."""
+        provider_cfg = LighthouseProviderConfiguration(
+            tenant_id=tenants_fixture[0].id,
+            provider_type=LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+            is_active=True,
+        )
+        # Store valid credentials first to pass model validation
+        provider_cfg.credentials_decoded = {
+            "api_key": "ABSKQmVkcm9ja0FQSUtleS" + ("A" * 110),
+            "region": "us-east-1",
+        }
+        provider_cfg.save()
+
+        # Mock the credentials_decoded property to return a non-dict value
+        # This simulates corrupted/invalid stored data
+        with patch.object(
+            type(provider_cfg),
+            "credentials_decoded",
+            new_callable=lambda: property(lambda self: "invalid"),
+        ):
+            result = _extract_bedrock_credentials(provider_cfg)
+
+        assert result is None
+
+
+class TestCreateBedrockClient:
+    """Unit tests for _create_bedrock_client helper function."""
+
+    @patch("tasks.jobs.lighthouse_providers.boto3.client")
+    def test_create_client_with_access_keys(self, mock_boto_client):
+        """Test creating client with access key authentication."""
+        mock_client = MagicMock()
+        mock_boto_client.return_value = mock_client
+
+        creds = {
+            "access_key_id": "AKIAIOSFODNN7EXAMPLE",
+            "secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+            "region": "us-east-1",
+        }
+
+        result = _create_bedrock_client(creds)
+
+        assert result == mock_client
+        mock_boto_client.assert_called_once_with(
+            service_name="bedrock",
+            region_name="us-east-1",
+            aws_access_key_id="AKIAIOSFODNN7EXAMPLE",
+            aws_secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+        )
+
+    @patch("tasks.jobs.lighthouse_providers.Config")
+    @patch("tasks.jobs.lighthouse_providers.boto3.client")
+    def test_create_client_with_api_key(self, mock_boto_client, mock_config):
+        """Test creating client with API key authentication."""
+        mock_client = MagicMock()
+        mock_events = MagicMock()
+        mock_client.meta.events = mock_events
+        mock_boto_client.return_value = mock_client
+        mock_config_instance = MagicMock()
+        mock_config.return_value = mock_config_instance
+        valid_api_key = "ABSKQmVkcm9ja0FQSUtleS" + ("A" * 110)
+
+        creds = {
+            "api_key": valid_api_key,
+            "region": "us-west-2",
+        }
+
+        result = _create_bedrock_client(creds)
+
+        assert result == mock_client
+        mock_boto_client.assert_called_once_with(
+            service_name="bedrock",
+            region_name="us-west-2",
+            config=mock_config_instance,
+        )
+        mock_events.register.assert_called_once()
+        call_args = mock_events.register.call_args
+        assert call_args[0][0] == "before-send.*.*"
+
+        # Verify handler injects bearer token
+        handler_fn = call_args[0][1]
+        mock_request = MagicMock()
+        mock_request.headers = {}
+        handler_fn(mock_request)
+        assert mock_request.headers["Authorization"] == f"Bearer {valid_api_key}"
+
+
 # TODO Move this to outputs/reports jobs
 @pytest.mark.django_db
 class TestGenerateOutputs:
@@ -1152,6 +1348,16 @@ class TestCheckLighthouseProviderConnectionTask:
                 None,
                 {"connected": True, "error": None},
             ),
+            # Bedrock API key authentication
+            (
+                LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+                {
+                    "api_key": "ABSKQmVkcm9ja0FQSUtleS" + ("A" * 110),
+                    "region": "us-east-1",
+                },
+                None,
+                {"connected": True, "error": None},
+            ),
         ],
     )
     def test_check_connection_success_all_providers(
@@ -1220,6 +1426,24 @@ class TestCheckLighthouseProviderConnectionTask:
                     "list_foundation_models",
                 ),
             ),
+            # Bedrock API key authentication failure
+            (
+                LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+                {
+                    "api_key": "ABSKQmVkcm9ja0FQSUtleS" + ("X" * 110),
+                    "region": "us-east-1",
+                },
+                None,
+                ClientError(
+                    {
+                        "Error": {
+                            "Code": "UnrecognizedClientException",
+                            "Message": "Invalid API key",
+                        }
+                    },
+                    "list_foundation_models",
+                ),
+            ),
         ],
     )
     def test_check_connection_api_failure(
@@ -1344,6 +1568,17 @@ class TestRefreshLighthouseProviderModelsTask:
                 {"openai.gpt-oss-120b-1:0": "gpt-oss-120b"},
                 1,
             ),
+            # Bedrock API key authentication
+            (
+                LighthouseProviderConfiguration.LLMProviderChoices.BEDROCK,
+                {
+                    "api_key": "ABSKQmVkcm9ja0FQSUtleS" + ("A" * 110),
+                    "region": "us-east-1",
+                },
+                None,
+                {"anthropic.claude-v3": "Claude 3"},
+                1,
+            ),
         ],
     )
     def test_refresh_models_create_new(