diff --git a/src/bedrock_agentcore/gateway/client.py b/src/bedrock_agentcore/gateway/client.py index 813ef540..e92ff0e5 100644 --- a/src/bedrock_agentcore/gateway/client.py +++ b/src/bedrock_agentcore/gateway/client.py @@ -381,6 +381,64 @@ def create_agentic_retrieve_target( **target_kwargs, ) + # Web Search target helpers + # ------------------------------------------------------------------------- + def create_web_search_target( + self, + gateway_identifier: str, + name: Optional[str] = None, + description: Optional[str] = None, + exclude_domains: Optional[List[str]] = None, + parameter_overrides: Optional[List[Dict[str, Any]]] = None, + wait_config: Optional[WaitConfig] = None, + **kwargs, + ) -> Dict[str, Any]: + """Create a gateway target that exposes Amazon Web Search as an MCP WebSearch tool. + + Args: + gateway_identifier: Gateway ID or ARN. + name: Target name. Defaults to "web-search". + description: Agent-facing description of the WebSearch tool. + exclude_domains: Optional list of domains to exclude from results. + parameter_overrides: Optional per-parameter visibility/description overrides. + wait_config: Optional WaitConfig for polling behavior. + **kwargs: Additional arguments forwarded to create_gateway_target + (e.g., credentialProviderConfigurations, roleArn). Overrides built values on conflict. + + Returns: + Gateway target details when READY. + """ + tool_config: Dict[str, Any] = {"name": "WebSearch"} + if exclude_domains: + tool_config["parameterValues"] = {"domainFilter": {"exclude": exclude_domains}} + if description: + tool_config["description"] = description + if parameter_overrides: + tool_config["parameterOverrides"] = parameter_overrides + + target_kwargs = { + "gatewayIdentifier": gateway_identifier, + "name": name or "web-search", + "targetConfiguration": { + "mcp": { + "connector": { + "source": {"connectorId": "web-search"}, + "enabled": ["WebSearch"], + "configurations": [tool_config], + }, + }, + }, + "credentialProviderConfigurations": [ + {"credentialProviderType": "GATEWAY_IAM_ROLE"}, + ], + } + target_kwargs.update(kwargs) + + return self.create_gateway_target_and_wait( + wait_config=wait_config, + **target_kwargs, + ) + # Name-based lookup # ------------------------------------------------------------------------- def get_gateway_by_name(self, name: str, **kwargs) -> Optional[Dict[str, Any]]: diff --git a/tests/unit/gateway/test_gateway_web_search_targets.py b/tests/unit/gateway/test_gateway_web_search_targets.py new file mode 100644 index 00000000..f220607e --- /dev/null +++ b/tests/unit/gateway/test_gateway_web_search_targets.py @@ -0,0 +1,125 @@ +"""Tests for GatewayClient Web Search target helper methods.""" + +from unittest.mock import MagicMock, Mock + +from bedrock_agentcore.gateway.client import GatewayClient + + +class TestCreateWebSearchTarget: + """Tests for create_web_search_target.""" + + def _make_client(self): + mock_session = MagicMock() + mock_session.region_name = "us-west-2" + client = GatewayClient(boto3_session=mock_session) + client.create_gateway_target_and_wait = Mock(return_value={"status": "READY", "targetId": "t-789"}) + return client + + def test_minimal(self): + client = self._make_client() + + result = client.create_web_search_target(gateway_identifier="gw-123") + + assert result["status"] == "READY" + client.create_gateway_target_and_wait.assert_called_once_with( + wait_config=None, + gatewayIdentifier="gw-123", + name="web-search", + targetConfiguration={ + "mcp": { + "connector": { + "source": {"connectorId": "web-search"}, + "enabled": ["WebSearch"], + "configurations": [{"name": "WebSearch"}], + }, + }, + }, + credentialProviderConfigurations=[ + {"credentialProviderType": "GATEWAY_IAM_ROLE"}, + ], + ) + + def test_with_all_options(self): + client = self._make_client() + + result = client.create_web_search_target( + gateway_identifier="gw-123", + name="custom-search", + description="Search the public web", + exclude_domains=["example.com", "spam.example"], + parameter_overrides=[{"path": "/maxResults", "visible": True}], + ) + + assert result["status"] == "READY" + call_kwargs = client.create_gateway_target_and_wait.call_args[1] + assert call_kwargs["name"] == "custom-search" + connector = call_kwargs["targetConfiguration"]["mcp"]["connector"] + assert connector["source"]["connectorId"] == "web-search" + assert connector["enabled"] == ["WebSearch"] + config = connector["configurations"][0] + assert config["name"] == "WebSearch" + assert config["description"] == "Search the public web" + assert config["parameterValues"] == {"domainFilter": {"exclude": ["example.com", "spam.example"]}} + assert config["parameterOverrides"] == [{"path": "/maxResults", "visible": True}] + + def test_no_parameter_values_when_no_exclude_domains(self): + client = self._make_client() + + client.create_web_search_target(gateway_identifier="gw-123") + + call_kwargs = client.create_gateway_target_and_wait.call_args[1] + config = call_kwargs["targetConfiguration"]["mcp"]["connector"]["configurations"][0] + assert "parameterValues" not in config + + def test_empty_exclude_domains_is_omitted(self): + client = self._make_client() + + client.create_web_search_target(gateway_identifier="gw-123", exclude_domains=[]) + + call_kwargs = client.create_gateway_target_and_wait.call_args[1] + config = call_kwargs["targetConfiguration"]["mcp"]["connector"]["configurations"][0] + assert "parameterValues" not in config + + def test_kwargs_override_target_configuration(self): + client = self._make_client() + + custom_target_config = {"mcp": {"lambda": {"lambdaArn": "arn:..."}}} + client.create_web_search_target( + gateway_identifier="gw-123", + targetConfiguration=custom_target_config, + ) + + call_kwargs = client.create_gateway_target_and_wait.call_args[1] + assert call_kwargs["targetConfiguration"] == custom_target_config + + def test_kwargs_override_credential_provider(self): + client = self._make_client() + + custom_creds = [{"credentialProviderType": "CUSTOM"}] + client.create_web_search_target( + gateway_identifier="gw-123", + credentialProviderConfigurations=custom_creds, + ) + + call_kwargs = client.create_gateway_target_and_wait.call_args[1] + assert call_kwargs["credentialProviderConfigurations"] == custom_creds + + def test_default_credential_provider(self): + client = self._make_client() + + client.create_web_search_target(gateway_identifier="gw-123") + + call_kwargs = client.create_gateway_target_and_wait.call_args[1] + assert call_kwargs["credentialProviderConfigurations"] == [ + {"credentialProviderType": "GATEWAY_IAM_ROLE"}, + ] + + def test_wait_config_passed_through(self): + from bedrock_agentcore._utils.config import WaitConfig + + client = self._make_client() + wc = WaitConfig(max_wait=60, poll_interval=5) + + client.create_web_search_target(gateway_identifier="gw-123", wait_config=wc) + + assert client.create_gateway_target_and_wait.call_args[1]["wait_config"] == wc diff --git a/tests_integ/gateway/test_gateway_web_search_targets.py b/tests_integ/gateway/test_gateway_web_search_targets.py new file mode 100644 index 00000000..41b113ee --- /dev/null +++ b/tests_integ/gateway/test_gateway_web_search_targets.py @@ -0,0 +1,103 @@ +"""Integration tests for GatewayClient Web Search target helper methods. + +Requires environment variables: + BEDROCK_TEST_REGION: AWS region (default: us-west-2) + GATEWAY_ROLE_ARN: IAM role ARN with AgentCore gateway trust policy +""" + +import os +import time + +import pytest +from botocore.exceptions import ClientError + +from bedrock_agentcore.gateway.client import GatewayClient + + +@pytest.mark.integration +class TestGatewayWebSearchTarget: + """Integration tests for create_web_search_target.""" + + @classmethod + def setup_class(cls): + cls.region = os.environ.get("BEDROCK_TEST_REGION", "us-west-2") + cls.gateway_role_arn = os.environ.get("GATEWAY_ROLE_ARN") + if not cls.gateway_role_arn: + pytest.fail("GATEWAY_ROLE_ARN must be set") + + cls.gateway_client = GatewayClient(region_name=cls.region) + cls.test_prefix = f"sdk-integ-ws-tgt-{int(time.time())}" + cls.gateway_id = None + cls.target_ids = [] + + gw = cls.gateway_client.create_gateway_and_wait( + name=f"{cls.test_prefix}-gw", + roleArn=cls.gateway_role_arn, + authorizerType="NONE", + protocolType="MCP", + ) + cls.gateway_id = gw["gatewayId"] + + @classmethod + def teardown_class(cls): + for target_id in cls.target_ids: + try: + cls.gateway_client.delete_gateway_target_and_wait( + gatewayIdentifier=cls.gateway_id, + targetId=target_id, + ) + except Exception as e: + print(f"Failed to delete target {target_id}: {e}") + + if cls.gateway_id: + try: + cls.gateway_client.delete_gateway_and_wait(gatewayIdentifier=cls.gateway_id) + except Exception as e: + print(f"Failed to delete gateway {cls.gateway_id}: {e}") + + def _create_target(self, **kwargs): + """Create a web search target, skipping the test if the account is not entitled. + + The web-search connector is enabled per account. When it is not, CreateGatewayTarget + rejects the request with "Connector integration web-search is not available for this + account." Any other error still fails the test. + """ + try: + return self.gateway_client.create_web_search_target(gateway_identifier=self.gateway_id, **kwargs) + except ClientError as e: + error = e.response.get("Error", {}) + if error.get("Code") == "ValidationException" and "not available for this account" in error.get( + "Message", "" + ): + pytest.skip(f"web-search connector not enabled for this account: {error.get('Message')}") + raise + + @pytest.mark.order(1) + def test_create_web_search_target_minimal(self): + target = self._create_target() + self.__class__.target_ids.append(target["targetId"]) + assert target["status"] == "READY" + assert target["name"] == "web-search" + + @pytest.mark.order(2) + def test_create_web_search_target_with_options(self): + target = self._create_target( + name=f"{self.test_prefix}-custom", + description="Search the public web", + exclude_domains=["example.com"], + parameter_overrides=[{"path": "/maxResults", "visible": True, "description": "How many results"}], + ) + self.__class__.target_ids.append(target["targetId"]) + assert target["status"] == "READY" + assert target["name"] == f"{self.test_prefix}-custom" + + @pytest.mark.order(3) + def test_create_web_search_target_with_credential_config(self): + target = self._create_target( + name=f"{self.test_prefix}-cred", + credentialProviderConfigurations=[ + {"credentialProviderType": "GATEWAY_IAM_ROLE"}, + ], + ) + self.__class__.target_ids.append(target["targetId"]) + assert target["status"] == "READY"