diff --git a/src/bedrock_agentcore/tools/integrations/__init__.py b/src/bedrock_agentcore/tools/integrations/__init__.py new file mode 100644 index 00000000..28608fdc --- /dev/null +++ b/src/bedrock_agentcore/tools/integrations/__init__.py @@ -0,0 +1 @@ +"""Framework integrations for the AgentCore built-in tools.""" diff --git a/src/bedrock_agentcore/tools/integrations/strands/README.md b/src/bedrock_agentcore/tools/integrations/strands/README.md new file mode 100644 index 00000000..01b53230 --- /dev/null +++ b/src/bedrock_agentcore/tools/integrations/strands/README.md @@ -0,0 +1,203 @@ +# AgentCore Web Search for Strands Agents + +`AgentCoreWebSearch` exposes the Amazon Bedrock AgentCore Web Search tool as a +[Strands Agents](https://strandsagents.com) tool, so an agent can answer questions about +recent events and cite its sources. + +## Overview + +- **No search API key** — calls are authenticated with SigV4 from the ambient AWS credentials, so there is no separate search provider account, key rotation or per-key billing +- **Source attribution** — every result carries a title, a URL, a publication date when the index reports one, and an extract, and the tool description asks the model to cite the URLs it used +- **Server-side filtering** — domain include and exclude lists and publication date bounds are applied by the service, not by trimming results afterwards +- **One tool, not a toolkit** — `instance.web_search` is a single Strands tool, so it drops into an existing `Agent(tools=[...])` list + +## How it works + +Web search reaches the service through an AgentCore Gateway connector target: + +``` +┌─────────┐ ┌──────────────────┐ ┌──────────────┐ ┌────────────────┐ +│ Agent │────▶│ AgentCoreWebSearch│────▶│ Gateway │────▶│ web-search │ +│ │ │ (SigV4, MCP) │ │ connector │ │ connector │ +│ │◀────│ formats results │◀────│ target │◀────│ (AWS managed) │ +└─────────┘ └──────────────────┘ └──────────────┘ └────────────────┘ + caller's credentials gateway execution role +``` + +The caller's credentials sign the request to the gateway. The gateway then uses its own +execution role to call the connector, so two different principals need permissions. See +[IAM](#iam) below. + +Which transport is used is decided inside `WebSearchClient`, not here. Every gateway +argument on this class is optional and only the ones actually supplied are forwarded, so a +later direct API needing none of them works through the same call. + +## Installation + +```bash +pip install 'bedrock-agentcore[strands-agents]' +``` + +## Prerequisites + +The web search connector is enabled per account and is offered in a subset of regions. A +gateway with a web search target has to exist before the tool can be used; creating one is +three calls and is done once per account: + +```python +from bedrock_agentcore.gateway.client import GatewayClient + +client = GatewayClient(region_name="us-east-1") + +gateway = client.create_gateway_and_wait( + name="my-web-search-gateway", + roleArn="arn:aws:iam::111122223333:role/MyGatewayExecutionRole", + authorizerType="AWS_IAM", + protocolType="MCP", +) + +client.create_web_search_target(gateway_identifier=gateway["gatewayId"]) +``` + +For the regions the connector is currently offered in, see +[Web Search Tool](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/gateway-target-connector-web-search-tool.html) +in the AgentCore developer guide. + +## Quick start + +```python +from strands import Agent + +from bedrock_agentcore.tools.integrations.strands import AgentCoreWebSearch + +with AgentCoreWebSearch(region="us-east-1", gateway_id="my-web-search-gateway-abc123") as search: + agent = Agent(tools=[search.web_search]) + agent("What changed in the most recent boto3 release?") +``` + +Without the context manager, call `close()` when the agent is done: + +```python +search = AgentCoreWebSearch(gateway_id="my-web-search-gateway-abc123") +agent = Agent(tools=[search.web_search]) +agent("Who maintains urllib3?") +search.close() +``` + +## Configuration + +| Argument | Description | +|---|---| +| `region` | AWS region to call. Defaults to `us-east-1`, except when `gateway_arn` is given, in which case the ARN's region is used. | +| `gateway_id` | ID of a gateway carrying a web search connector target. | +| `gateway_arn` | ARN of that gateway. The ID and the region are read from it. | +| `gateway_endpoint` | A gateway MCP endpoint URL, if one is already known. | +| `target_name` | Name the connector target was created under. Supplying it avoids a tool discovery round trip on the first search. | +| `tool_name` | Fully qualified name of the gateway tool, if already known. | +| `boto3_session` | Session to take credentials from. | +| `client` | A `WebSearchClient` to use as is. The caller keeps ownership of it, so `close()` leaves it open. | + +Exactly one of `gateway_id`, `gateway_arn`, `gateway_endpoint` or `client` says where the +search goes. + +`region` defaults to `us-east-1` rather than to the session's region, because web search is +not offered everywhere. It is filled in only when no region and no gateway ARN were given: +an explicit region takes precedence over the one in a gateway ARN, so defaulting it +unconditionally would send a `eu-west-1` gateway's traffic to `us-east-1`. + +## Tool arguments + +The model sees one tool, `web_search`. Only `query` is required. + +| Argument | Description | +|---|---| +| `query` | What to search for, as a natural language query. 200 characters or fewer, checked locally. | +| `max_results` | How many results to return, between 1 and 25. Unset means the service default of 10. | +| `include_domains` | Only return results from these domains, up to 100. A root domain also matches its subdomains. | +| `exclude_domains` | Drop results from these domains, up to 100. | +| `published_after` | Only return pages published on or after this ISO-8601 UTC timestamp, inclusive. Applies to web results only. | +| `published_before` | Only return pages published on or before this one, inclusive. Applies to web results only. | + +Request filters compose with the target's own domain rules and can never widen them. A +domain is returned only if it appears on every include list that is set, so when the target +was created with an include list, a request-level `include_domains` narrows to the +intersection of the two. If the two share no domains the search returns nothing, and that is +a silent empty result rather than an error. + +Target-level lists are applied server side and are not visible to the model, so a model +asking repeatedly for a domain the target excludes gets nothing back and cannot tell why. +Keeping the restriction in the tool arguments instead is what lets it see the boundary it is +working inside. + +Request-level filter arguments need connector version 1.2.0 or later on the target. On an +earlier version the tool accepts only `query` and `max_results`. + +## IAM + +Two permissions on two different principals, and getting this wrong is the most common +setup failure. + +The caller's credentials need `bedrock-agentcore:InvokeGateway` on the gateway: + +```json +{ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "bedrock-agentcore:InvokeGateway", + "Resource": "arn:aws:bedrock-agentcore:us-east-1:111122223333:gateway/my-gateway-abc123" + }] +} +``` + +The gateway's execution role needs `bedrock-agentcore:InvokeWebSearch` on the connector, and +nothing else: + +```json +{ + "Version": "2012-10-17", + "Statement": [{ + "Effect": "Allow", + "Action": "bedrock-agentcore:InvokeWebSearch", + "Resource": "arn:aws:bedrock-agentcore:us-east-1:aws:tool/web-search.v1" + }] +} +``` + +Note the `aws` account field in that resource ARN: the connector is service owned, not +account owned. + +If the execution role's trust policy narrows `aws:SourceArn` to a gateway in a different +region than the gateway actually created, `CreateGateway` still succeeds and the first +search fails with `Failed to obtain execution role credentials`, which reads like an +entitlement problem rather than an IAM one. + +## Errors + +A failed search raises rather than returning the message as a result, because a string +saying the search failed is indistinguishable to the model from a search that found nothing: + +- `WebSearchError` when the search itself fails, including when the account is not entitled to the connector (`not available for this account`). An unsupported region reports the same message, so check the region before concluding the account lacks the entitlement. +- `ValueError` when an argument is outside the documented limits, or the tool has already been closed. + +Neither reaches the caller of `agent(...)`. The Strands tool executor turns a tool +exception into a tool result with `status` of `error` carrying the message, so the agent +keeps running and the model sees that its search failed rather than that the web is empty. +The distinction is the point: an error result is something a model can react to, a +successful result reading "search failed" is not. + +An empty result set is not an error. It returns a sentence saying so and naming an +over-narrow filter as the likely cause, so the model can widen the query itself. When a +filtered search keeps coming back empty, check the target's own domain configuration before +the query: a request include list that shares no domains with the target's include list +returns nothing, silently. + +## Telemetry + +Calls are attributed to Strands through `integration_source="strands"` on the user agent, so +web search usage from Strands agents is distinguishable from usage through the raw SDK. + +## Related + +- [`WebSearchClient`](../../web_search_client.py) — the underlying client, usable without a framework +- [Web Search Tool](https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/gateway-target-connector-web-search-tool.html) — the service documentation diff --git a/src/bedrock_agentcore/tools/integrations/strands/__init__.py b/src/bedrock_agentcore/tools/integrations/strands/__init__.py new file mode 100644 index 00000000..90f9cf90 --- /dev/null +++ b/src/bedrock_agentcore/tools/integrations/strands/__init__.py @@ -0,0 +1,5 @@ +"""Strands Agents framework integration for the AgentCore built-in tools.""" + +from .web_search import DEFAULT_REGION, AgentCoreWebSearch + +__all__ = ["AgentCoreWebSearch", "DEFAULT_REGION"] diff --git a/src/bedrock_agentcore/tools/integrations/strands/web_search.py b/src/bedrock_agentcore/tools/integrations/strands/web_search.py new file mode 100644 index 00000000..b2b81833 --- /dev/null +++ b/src/bedrock_agentcore/tools/integrations/strands/web_search.py @@ -0,0 +1,237 @@ +"""AgentCore Web Search as a Strands Agents tool. + +Web search reaches the service through an AgentCore Gateway connector target +today, and a direct API is expected later. Which transport is used is decided +inside ``WebSearchClient`` rather than here, so every gateway argument on this +class is keyword-only and optional, and only the arguments actually supplied are +forwarded. A later transport needing none of them works through the same call, +without a signature change here. +""" + +import logging +import threading +from typing import Any, Dict, List, Optional + +from strands.tools import tool + +from ...web_search_client import WebSearchClient, WebSearchResponse + +logger = logging.getLogger(__name__) + +#: Web search is offered in a subset of regions, so this default is not the one +#: the browser and code interpreter tools use. It is applied only when no region +#: and no gateway ARN were given, because the client prefers an explicit region +#: over the one carried in the ARN. +DEFAULT_REGION = "us-east-1" + + +def _format_response(response: WebSearchResponse) -> str: + """Render search results as text a model can cite from. + + Args: + response: The results of one search. + + Returns: + One numbered block per result, or a plain sentence when there were none. + """ + if not response.results: + return "No results. The query may be too narrow, or a domain or date filter may have excluded everything." + + blocks: List[str] = [] + for position, result in enumerate(response.results, start=1): + lines = [f"{position}. {result.title or 'Untitled'}"] + if result.url: + lines.append(f" URL: {result.url}") + if result.published_date: + lines.append(f" Published: {result.published_date}") + if result.text: + lines.append(f" {result.text}") + blocks.append("\n".join(lines)) + return "\n\n".join(blocks) + + +class AgentCoreWebSearch: + """Exposes AgentCore Web Search as a Strands tool. + + The search runs over an AgentCore Gateway target that has the web search + connector attached. Calls are authenticated with SigV4 from the ambient AWS + credentials; there is no web search API key. Those credentials need + ``bedrock-agentcore:InvokeGateway`` on the gateway, and the gateway's own + service role needs ``bedrock-agentcore:InvokeWebSearch`` on the connector. + + Basic Usage: + >>> from strands import Agent + >>> from bedrock_agentcore.tools.integrations.strands import AgentCoreWebSearch + >>> + >>> search = AgentCoreWebSearch(region="us-east-1", gateway_id="my-gateway-abc123") + >>> agent = Agent(tools=[search.web_search]) + >>> agent("What changed in the most recent boto3 release?") + >>> search.close() + + Context Manager: + >>> with AgentCoreWebSearch(gateway_id="my-gateway-abc123") as search: + ... agent = Agent(tools=[search.web_search]) + ... agent("Who maintains urllib3?") + """ + + def __init__( + self, + region: Optional[str] = None, + *, + gateway_id: Optional[str] = None, + gateway_arn: Optional[str] = None, + gateway_endpoint: Optional[str] = None, + target_name: Optional[str] = None, + tool_name: Optional[str] = None, + boto3_session: Optional[Any] = None, + client: Optional[WebSearchClient] = None, + ): + """Initialize the tool. + + Exactly one of ``gateway_id``, ``gateway_arn``, ``gateway_endpoint`` or + ``client`` says where the search goes. + + Args: + region: AWS region to call. Defaults to ``DEFAULT_REGION``, except when + ``gateway_arn`` is given, in which case the ARN's region is used. + gateway_id: ID of a gateway carrying a web search connector target. + gateway_arn: ARN of that gateway. The ID and the region are read from it. + gateway_endpoint: A gateway MCP endpoint URL, if one is already known. + target_name: Name the connector target was created under. Supplying it + avoids a tool discovery round trip on the first search. + tool_name: Fully qualified name of the gateway tool, if already known. + boto3_session: Session to take credentials from. + client: A client to use as is. The caller keeps ownership of it, so + ``close`` leaves it open. + + Raises: + ValueError: If both ``client`` and a gateway argument are given. + """ + if client is not None and any(value is not None for value in (gateway_id, gateway_arn, gateway_endpoint)): + raise ValueError("Pass either client or one of gateway_id, gateway_arn or gateway_endpoint, not both.") + + # WebSearchClient prefers an explicit region over the one in a gateway + # ARN, so filling the default in here would silently override the ARN's + # region. Leaving it unset is what lets the ARN decide. + if region is None and gateway_arn is None: + region = DEFAULT_REGION + + self.region = region + self._gateway_id = gateway_id + self._gateway_arn = gateway_arn + self._gateway_endpoint = gateway_endpoint + self._target_name = target_name + self._tool_name = tool_name + self._boto3_session = boto3_session + self._owns_client = client is None + self._client = client if client is not None else self._create_client() + self._closed = False + self._lock = threading.Lock() + + def _create_client(self) -> WebSearchClient: + """Build the client, forwarding only the arguments that were given. + + Returns: + A client pointed at whichever destination was identified. + """ + # Passing gateway_id=None explicitly would tie this module to the gateway + # transport. Forwarding only what was supplied means a future transport + # needing no gateway argument works through this same path. + kwargs: Dict[str, Any] = {"integration_source": "strands"} + optional = ( + ("region", self.region), + ("gateway_id", self._gateway_id), + ("gateway_arn", self._gateway_arn), + ("gateway_endpoint", self._gateway_endpoint), + ("target_name", self._target_name), + ("tool_name", self._tool_name), + ("boto3_session", self._boto3_session), + ) + for name, value in optional: + if value is not None: + kwargs[name] = value + return WebSearchClient(**kwargs) + + @tool + def web_search( + self, + query: str, + max_results: Optional[int] = None, + include_domains: Optional[List[str]] = None, + exclude_domains: Optional[List[str]] = None, + published_after: Optional[str] = None, + published_before: Optional[str] = None, + ) -> str: + """Search the web for current information and return source-attributed results. + + Use this tool for questions about recent events, releases or prices, for + facts that need a citable source, and for topics outside your training + data. Each result carries a title, a URL, a publication date when the + index reports one, and an extract. Cite the URLs you used in your answer. + + Prefer one broad query over several narrow ones. Use include_domains to + restrict a search to sources you trust, such as official documentation. + + Args: + query: What to search for, as a natural language query. 200 characters or fewer. + max_results: How many results to return, between 1 and 25. Leave unset + for the service default. + include_domains: Only return results from these domains, e.g. + ["docs.aws.amazon.com"]. A root domain also matches its subdomains. + Can only narrow the search, never widen it. + exclude_domains: Drop results from these domains. + published_after: Only return pages published on or after this date, as + ISO-8601 UTC, e.g. 2026-01-01T00:00:00Z. + published_before: Only return pages published on or before this date, + as ISO-8601 UTC. + + Returns: + One numbered block per result, carrying the title, URL, publication + date and an extract. + + Raises: + ValueError: If the tool has been closed, or an argument is outside the + documented limits. + WebSearchError: If the search fails. + """ + with self._lock: + if self._closed: + raise ValueError("This AgentCoreWebSearch has been closed.") + client = self._client + + try: + response = client.search( + query, + max_results=max_results, + include_domains=include_domains, + exclude_domains=exclude_domains, + published_after=published_after, + published_before=published_before, + ) + except Exception as exc: + logger.error("Web search failed: %s - %s", type(exc).__name__, exc) + raise + + logger.debug("Web search returned %d results", len(response.results)) + return _format_response(response) + + def close(self) -> None: + """Release the underlying client, if this object created it.""" + with self._lock: + if self._closed: + return + self._closed = True + if self._owns_client: + self._client.close() + + def __enter__(self) -> "AgentCoreWebSearch": + """Enter the context manager. + + Returns: + This object. + """ + return self + + def __exit__(self, *exc_info: Any) -> None: + """Close on exit.""" + self.close() diff --git a/tests/bedrock_agentcore/tools/integrations/__init__.py b/tests/bedrock_agentcore/tools/integrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bedrock_agentcore/tools/integrations/strands/__init__.py b/tests/bedrock_agentcore/tools/integrations/strands/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py b/tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py new file mode 100644 index 00000000..800d9e74 --- /dev/null +++ b/tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py @@ -0,0 +1,222 @@ +"""Tests for AgentCoreWebSearch.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from bedrock_agentcore.tools.integrations.strands.web_search import ( + DEFAULT_REGION, + AgentCoreWebSearch, + _format_response, +) +from bedrock_agentcore.tools.web_search_client import WebSearchError, WebSearchResponse, WebSearchResult + +ARN = "arn:aws:bedrock-agentcore:eu-west-1:111122223333:gateway/my-gw-abc123" + + +def _response(*results): + return WebSearchResponse(results=list(results), search_id="search-1") + + +def _result(**kwargs): + fields = {"text": "", "url": None, "title": None, "published_date": None} + fields.update(kwargs) + return WebSearchResult(**fields) + + +@pytest.fixture +def client(): + """A stand-in WebSearchClient whose searches return one result.""" + client = MagicMock() + client.search.return_value = _response(_result(title="t", url="https://example.com", text="body")) + return client + + +class TestClientConstruction: + """How the underlying WebSearchClient gets built.""" + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_leaves_region_unset_when_a_gateway_arn_carries_one(self, mock_client): + AgentCoreWebSearch(gateway_arn=ARN) + + # Filling in a default here would send a eu-west-1 gateway to us-east-1, + # because the client prefers an explicit region over the ARN's. + assert "region" not in mock_client.call_args.kwargs + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_an_explicit_region_still_wins_over_a_gateway_arn(self, mock_client): + AgentCoreWebSearch("us-east-1", gateway_arn=ARN) + + assert mock_client.call_args.kwargs["region"] == "us-east-1" + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_defaults_the_region_when_no_arn_is_given(self, mock_client): + AgentCoreWebSearch(gateway_id="my-gw-abc123") + + assert mock_client.call_args.kwargs["region"] == DEFAULT_REGION + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_forwards_only_the_arguments_that_were_given(self, mock_client): + AgentCoreWebSearch(gateway_id="my-gw-abc123") + + # An explicit gateway_id=None would tie this module to the gateway + # transport, so absent arguments must not be forwarded at all. + assert set(mock_client.call_args.kwargs) == {"integration_source", "region", "gateway_id"} + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_attributes_usage_to_strands(self, mock_client): + AgentCoreWebSearch(gateway_id="my-gw-abc123") + + assert mock_client.call_args.kwargs["integration_source"] == "strands" + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_forwards_the_discovery_shortcuts(self, mock_client): + AgentCoreWebSearch(gateway_id="my-gw-abc123", target_name="amazon-web-search", tool_name="x___WebSearch") + + assert mock_client.call_args.kwargs["target_name"] == "amazon-web-search" + assert mock_client.call_args.kwargs["tool_name"] == "x___WebSearch" + + def test_rejects_a_client_alongside_a_gateway_argument(self, client): + with pytest.raises(ValueError, match="not both"): + AgentCoreWebSearch(gateway_id="my-gw-abc123", client=client) + + +class TestToolSpec: + """The tool spec Strands derives from the decorated method.""" + + def test_is_named_web_search(self, client): + assert AgentCoreWebSearch(client=client).web_search.tool_name == "web_search" + + def test_takes_the_query_plus_every_filter_and_requires_only_the_query(self, client): + schema = AgentCoreWebSearch(client=client).web_search.tool_spec["inputSchema"]["json"] + + assert schema["required"] == ["query"] + assert set(schema["properties"]) == { + "query", + "max_results", + "include_domains", + "exclude_domains", + "published_after", + "published_before", + } + + def test_asks_the_model_to_cite_its_sources(self, client): + # Citing sources is a condition of use for this connector, so losing this + # sentence from the description is a compliance problem, not a wording one. + description = AgentCoreWebSearch(client=client).web_search.tool_spec["description"] + + assert "Cite the URLs" in description + + +class TestSearching: + """What reaches the client, and what comes back.""" + + def test_passes_the_query_through(self, client): + AgentCoreWebSearch(client=client).web_search("who maintains urllib3") + + assert client.search.call_args.args == ("who maintains urllib3",) + + def test_forwards_every_filter(self, client): + AgentCoreWebSearch(client=client).web_search( + "python releases", + max_results=5, + include_domains=["python.org"], + exclude_domains=["spam.example"], + published_after="2026-01-01T00:00:00Z", + published_before="2026-06-01T00:00:00Z", + ) + + assert client.search.call_args.kwargs == { + "max_results": 5, + "include_domains": ["python.org"], + "exclude_domains": ["spam.example"], + "published_after": "2026-01-01T00:00:00Z", + "published_before": "2026-06-01T00:00:00Z", + } + + def test_returns_the_formatted_results(self, client): + result = AgentCoreWebSearch(client=client).web_search("anything") + + assert "1. t" in result + assert "https://example.com" in result + + def test_raises_rather_than_returning_the_failure_as_text(self, client): + # Returning the message as a successful result would make a failed search + # look like a search that found nothing. + client.search.side_effect = WebSearchError("gateway said no") + + with pytest.raises(WebSearchError, match="gateway said no"): + AgentCoreWebSearch(client=client).web_search("anything") + + def test_refuses_to_search_once_closed(self, client): + search = AgentCoreWebSearch(client=client) + search.close() + + with pytest.raises(ValueError, match="closed"): + search.web_search("anything") + + +class TestFormatting: + """How results are rendered for the model.""" + + def test_numbers_each_result(self): + text = _format_response(_response(_result(title="first"), _result(title="second"))) + + assert text.startswith("1. first") + assert "2. second" in text + + def test_shows_the_publication_date(self): + # Recency is most of why an agent searches, so the date has to reach the model + # rather than being dropped on the way through. + text = _format_response(_response(_result(title="t", published_date="2026-09-14T00:00:00Z"))) + + assert "Published: 2026-09-14T00:00:00Z" in text + + def test_omits_fields_the_index_did_not_report(self): + text = _format_response(_response(_result(text="just an extract"))) + + assert "URL:" not in text + assert "Published:" not in text + assert "just an extract" in text + + def test_names_untitled_results(self): + assert "1. Untitled" in _format_response(_response(_result(url="https://example.com"))) + + def test_explains_an_empty_result_set(self): + # An empty list is what an over-narrow filter returns, and the model + # cannot tell that from a failure unless the text says so. + text = _format_response(_response()) + + assert "No results" in text + assert "filter" in text + + +class TestLifecycle: + """Ownership of the client.""" + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_closes_a_client_it_created(self, mock_client): + search = AgentCoreWebSearch(gateway_id="my-gw-abc123") + search.close() + + mock_client.return_value.close.assert_called_once() + + def test_leaves_an_injected_client_open(self, client): + AgentCoreWebSearch(client=client).close() + + client.close.assert_not_called() + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_closing_twice_closes_once(self, mock_client): + search = AgentCoreWebSearch(gateway_id="my-gw-abc123") + search.close() + search.close() + + mock_client.return_value.close.assert_called_once() + + @patch("bedrock_agentcore.tools.integrations.strands.web_search.WebSearchClient") + def test_closes_on_leaving_a_with_block(self, mock_client): + with AgentCoreWebSearch(gateway_id="my-gw-abc123") as search: + assert search is not None + + mock_client.return_value.close.assert_called_once() diff --git a/tests_integ/tools/integrations/__init__.py b/tests_integ/tools/integrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests_integ/tools/integrations/strands/test_web_search_integration.py b/tests_integ/tools/integrations/strands/test_web_search_integration.py new file mode 100644 index 00000000..4c73007f --- /dev/null +++ b/tests_integ/tools/integrations/strands/test_web_search_integration.py @@ -0,0 +1,159 @@ +"""Integration tests for AgentCoreWebSearch. + +These call a real gateway that already has a web search connector target on it. The +connector is enabled per account, so they skip rather than fail when the account is not +entitled to it. + +Run with: + uv run pytest tests_integ/tools/integrations/strands/test_web_search_integration.py -xvs + +Requires environment variables: + WEB_SEARCH_GATEWAY_ID: ID of a gateway with a web search connector target + BEDROCK_TEST_REGION: AWS region (default: us-east-1). The connector is offered in + us-east-1, eu-west-1 and ap-northeast-1. + WEB_SEARCH_TARGET_NAME: Optional. Name of the web search target on that gateway. + Supplying it saves a tools/list call, since Gateway prefixes every tool with the + name of the target it came from. + STRANDS_TEST_MODEL_ID: Optional. Model to run the agent test against. + Default: us.anthropic.claude-sonnet-4-5-20250929-v1:0. That test also needs + bedrock:InvokeModel on the model and skips if the model is not reachable. +""" + +import os + +import pytest +from strands import Agent + +from bedrock_agentcore.tools.integrations.strands import AgentCoreWebSearch +from bedrock_agentcore.tools.web_search_client import WebSearchError + +DEFAULT_MODEL_ID = "us.anthropic.claude-sonnet-4-5-20250929-v1:0" + + +@pytest.mark.integration +class TestAgentCoreWebSearchIntegration: + """AgentCoreWebSearch against a live gateway target.""" + + @classmethod + def setup_class(cls): + cls.gateway_id = os.environ.get("WEB_SEARCH_GATEWAY_ID") + if not cls.gateway_id: + pytest.skip("WEB_SEARCH_GATEWAY_ID must be set") + cls.region = os.environ.get("BEDROCK_TEST_REGION", "us-east-1") + cls.target_name = os.environ.get("WEB_SEARCH_TARGET_NAME") + cls.model_id = os.environ.get("STRANDS_TEST_MODEL_ID", DEFAULT_MODEL_ID) + + def _search_tool(self): + return AgentCoreWebSearch( + region=self.region, + gateway_id=self.gateway_id, + target_name=self.target_name, + ) + + def _search(self, tool, query, **kwargs): + """Search, skipping when the account is not entitled to the connector.""" + try: + return tool.web_search(query, **kwargs) + except WebSearchError as e: + if "not available for this account" in str(e): + pytest.skip(f"web-search connector not enabled for this account: {e}") + raise + + def test_search_returns_citable_results(self): + with self._search_tool() as tool: + text = self._search(tool, "what is amazon bedrock agentcore", max_results=3) + + assert text.startswith("1. ") + # Citations must be retained for any output shown to an end user, so a URL has + # to survive into the text the model reads. + assert "URL: http" in text + assert "No results" not in text + + def test_search_respects_max_results(self): + with self._search_tool() as tool: + text = self._search(tool, "python urllib3 release notes", max_results=2) + + assert "3. " not in text + + def test_search_with_domain_filter(self): + """Needs connector version 1.2.0 or later on the target. + + A request-level include filter is only accepted from 1.2.0 on, and the version of + an existing gateway's target is not this test's to choose, so an older target + skips rather than fails. + """ + with self._search_tool() as tool: + try: + text = self._search( + tool, + "agentcore gateway connector targets", + max_results=5, + include_domains=["docs.aws.amazon.com"], + ) + except WebSearchError as e: + if "domainFilter" in str(e) or "include" in str(e): + pytest.skip(f"target's connector version does not accept an include filter: {e}") + raise + + if "No results" in text: + pytest.skip("include filter returned nothing; check the target's own domain rules") + for line in text.splitlines(): + if line.strip().startswith("URL: "): + assert "aws.amazon.com" in line + + def test_tool_name_discovery_without_a_target_name(self): + """Without target_name the client finds the tool through tools/list.""" + with AgentCoreWebSearch(region=self.region, gateway_id=self.gateway_id) as tool: + self._search(tool, "bedrock agentcore gateway", max_results=1) + + assert tool._client.backend._tool_name.endswith("WebSearch") + + def test_the_client_is_reused_across_searches(self): + with self._search_tool() as tool: + self._search(tool, "first query", max_results=1) + self._search(tool, "second query", max_results=1) + + assert tool._client.backend._mcp_session_id + + def test_an_agent_calls_the_tool_and_cites_a_url(self): + """The end to end path: a model decides to search, and the URL reaches its answer. + + This is the only test here that exercises the Strands tool contract rather than + the method. It asserts a URL appears in the tool result recorded in the agent's + messages, not that the model's prose cites one, because the model's wording is + not this test's to assert. + + A search failure does not propagate out of ``agent(...)``: the Strands executor + turns a tool exception into a ``status: "error"`` tool result, which is why the + entitlement check reads the results rather than catching. A model that cannot be + reached does propagate, since that is not a tool call. + + The model is resolved by Strands through ``BedrockModel``, so it uses the ambient + AWS region rather than ``BEDROCK_TEST_REGION``. Those two are independent: web + search is offered in fewer regions than the models are. + """ + with self._search_tool() as tool: + agent = Agent(model=self.model_id, tools=[tool.web_search]) + try: + agent("Search the web for the newest boto3 release and give me the source URL.") + except Exception as e: + if "AccessDenied" in str(e) or "ValidationException" in str(e): + pytest.skip(f"test model {self.model_id} is not reachable: {e}") + raise + + tool_results = [ + block["toolResult"] + for message in agent.messages + for block in message.get("content", []) + if isinstance(block, dict) and "toolResult" in block + ] + + assert tool_results, "the model did not call the search tool" + returned = str(tool_results) + if "not available for this account" in returned: + pytest.skip(f"web-search connector not enabled for this account: {returned}") + + # A raised search failure arrives here as status error, so this also pins down + # that the failure stays distinguishable from a search that found nothing. + assert all(result.get("status") != "error" for result in tool_results), returned + assert "URL: http" in returned