From c42eb68868640d0a695e223493b5137ac7134fcd Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Mon, 21 Sep 2026 17:27:36 -0700 Subject: [PATCH] feat(tools): add AgentCore Web Search as a Strands Agents tool Exposes the web search tool as a single Strands tool, alongside the existing memory, payments and gateway Strands integrations, behind the same strands-agents extra. Attributed with integration_source="strands". Which transport is used stays inside WebSearchClient. Every gateway argument here is optional and only the ones supplied are forwarded, so the direct web search API will work through the same call without a signature change. Two behaviours the tests pin down: - region is defaulted only when no region and no gateway ARN were given. The client prefers an explicit region over the ARN's, so defaulting it unconditionally sends a eu-west-1 gateway's traffic to us-east-1. - a failed search raises rather than returning the message as a result. The Strands executor turns that into a status=error tool result, which a model can react to; a successful result reading "search failed" is indistinguishable to it from a search that found nothing. Unit tests cover the module to 100% branch coverage. The integration tests need a gateway with a web search target and have not been run. --- .../tools/integrations/__init__.py | 1 + .../tools/integrations/strands/README.md | 203 +++++++++++++++ .../tools/integrations/strands/__init__.py | 5 + .../tools/integrations/strands/web_search.py | 237 ++++++++++++++++++ .../tools/integrations/__init__.py | 0 .../tools/integrations/strands/__init__.py | 0 .../integrations/strands/test_web_search.py | 222 ++++++++++++++++ tests_integ/tools/integrations/__init__.py | 0 .../strands/test_web_search_integration.py | 159 ++++++++++++ 9 files changed, 827 insertions(+) create mode 100644 src/bedrock_agentcore/tools/integrations/__init__.py create mode 100644 src/bedrock_agentcore/tools/integrations/strands/README.md create mode 100644 src/bedrock_agentcore/tools/integrations/strands/__init__.py create mode 100644 src/bedrock_agentcore/tools/integrations/strands/web_search.py create mode 100644 tests/bedrock_agentcore/tools/integrations/__init__.py create mode 100644 tests/bedrock_agentcore/tools/integrations/strands/__init__.py create mode 100644 tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py create mode 100644 tests_integ/tools/integrations/__init__.py create mode 100644 tests_integ/tools/integrations/strands/test_web_search_integration.py 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