From c42eb68868640d0a695e223493b5137ac7134fcd Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Mon, 21 Sep 2026 17:27:36 -0700 Subject: [PATCH 1/4] 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 From cc6ab9eaaa585bef05d38cf00f8205d1ce0fd615 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Mon, 21 Sep 2026 18:38:01 -0700 Subject: [PATCH 2/4] docs(tools): align the web search flow diagram in the Strands README The AgentCoreWebSearch box's content lines were one column wider than its border, which skewed every box to its right. All five lines are now the same width with the edges in the same columns. --- .../tools/integrations/strands/README.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/bedrock_agentcore/tools/integrations/strands/README.md b/src/bedrock_agentcore/tools/integrations/strands/README.md index 01b53230..c25df68e 100644 --- a/src/bedrock_agentcore/tools/integrations/strands/README.md +++ b/src/bedrock_agentcore/tools/integrations/strands/README.md @@ -16,12 +16,12 @@ recent events and cite its sources. 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 +┌─────────┐ ┌────────────────────┐ ┌──────────────┐ ┌────────────────┐ +│ 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 From d2670ee895ed21e13b3ffc2b59c2ee3a6a11a68d Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Mon, 21 Sep 2026 18:51:18 -0700 Subject: [PATCH 3/4] test(tools): drive the Strands web search tool through a real agent loop The existing tests call web_search directly, so they verify the wrapper's wiring but not that Strands accepts the tool. These three drive a real Agent, with a scripted model standing in for the model's choice and the WebSearchClient stubbed, leaving the agent, the tool registry, the spec generation and the tool executor genuine. That pins down three things the direct tests cannot: the model's toolUse input binds to the method signature including the filter arguments, the formatted results arrive in the conversation as a success result, and a raised search failure becomes a status=error result the model can read while the agent keeps running. The last one was previously asserted in the README on the strength of reading the Strands executor source. --- .../integrations/strands/test_web_search.py | 85 +++++++++++++++++++ 1 file changed, 85 insertions(+) diff --git a/tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py b/tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py index 800d9e74..039da499 100644 --- a/tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py +++ b/tests/bedrock_agentcore/tools/integrations/strands/test_web_search.py @@ -1,8 +1,11 @@ """Tests for AgentCoreWebSearch.""" +import json from unittest.mock import MagicMock, patch import pytest +from strands import Agent +from strands.models import Model from bedrock_agentcore.tools.integrations.strands.web_search import ( DEFAULT_REGION, @@ -32,6 +35,50 @@ def client(): return client +class ScriptedModel(Model): + """A model that calls web_search once with fixed input, then answers. + + Only the model is scripted. The agent, its tool registry and its tool + executor are the real ones, which is the point of using this over a mock. + """ + + def __init__(self, tool_input): + self.tool_input = tool_input + self.calls = 0 + + def get_config(self): + return {} + + def update_config(self, **kwargs): + pass + + async def structured_output(self, output_model, prompt, **kwargs): + raise NotImplementedError + + async def stream(self, messages, tool_specs=None, system_prompt=None, **kwargs): + self.calls += 1 + if self.calls == 1: + yield {"messageStart": {"role": "assistant"}} + yield {"contentBlockStart": {"start": {"toolUse": {"toolUseId": "tu-1", "name": "web_search"}}}} + yield {"contentBlockDelta": {"delta": {"toolUse": {"input": json.dumps(self.tool_input)}}}} + yield {"contentBlockStop": {}} + yield {"messageStop": {"stopReason": "tool_use"}} + else: + yield {"messageStart": {"role": "assistant"}} + yield {"contentBlockDelta": {"delta": {"text": "done"}}} + yield {"contentBlockStop": {}} + yield {"messageStop": {"stopReason": "end_turn"}} + + +def _tool_results(agent): + return [ + block["toolResult"] + for message in agent.messages + for block in message.get("content", []) + if isinstance(block, dict) and "toolResult" in block + ] + + class TestClientConstruction: """How the underlying WebSearchClient gets built.""" @@ -156,6 +203,44 @@ def test_refuses_to_search_once_closed(self, client): search.web_search("anything") +class TestInsideAnAgentLoop: + """The tool driven through a real Agent, rather than called directly.""" + + def test_the_model_can_call_it_and_the_arguments_arrive(self, client): + search = AgentCoreWebSearch(client=client) + model = ScriptedModel({"query": "boto3 release notes", "max_results": 3, "include_domains": ["github.com"]}) + + Agent(model=model, tools=[search.web_search])("what changed in boto3?") + + assert client.search.call_args.args == ("boto3 release notes",) + assert client.search.call_args.kwargs["max_results"] == 3 + assert client.search.call_args.kwargs["include_domains"] == ["github.com"] + + def test_the_results_reach_the_conversation(self, client): + search = AgentCoreWebSearch(client=client) + agent = Agent(model=ScriptedModel({"query": "anything"}), tools=[search.web_search]) + + agent("what changed in boto3?") + + result = _tool_results(agent)[0] + assert result["status"] == "success" + assert "https://example.com" in result["content"][0]["text"] + + def test_a_failed_search_becomes_an_error_result_the_model_can_read(self, client): + # Raising is deliberate, and this is what it buys: the executor turns the + # exception into status=error, so the failure stays distinguishable from a + # search that found nothing, and the agent still gets to respond. + client.search.side_effect = WebSearchError("connector not available for this account") + search = AgentCoreWebSearch(client=client) + agent = Agent(model=ScriptedModel({"query": "anything"}), tools=[search.web_search]) + + agent("what changed in boto3?") + + result = _tool_results(agent)[0] + assert result["status"] == "error" + assert "connector not available for this account" in result["content"][0]["text"] + + class TestFormatting: """How results are rendered for the model.""" From a094cc2f9bfaee65680f7ad5360dfc9efd314545 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Mon, 21 Sep 2026 19:02:52 -0700 Subject: [PATCH 4/4] test(tools): fix two wrong assertions in the web search integ tests Both failed on their first live run against a gateway, and both were the test's fault rather than the tool's. max_results is honoured exactly, 1 gives 1 and 2 gives 2, but asserting "3. " is absent searches the result extracts too, and web prose contains a digit followed by a period often enough that this was never going to hold. Count the lines that open a numbered block instead. Mcp-Session-Id is optional in MCP and this connector does not send one, so the backend's session id stays None however many searches run through it. Assert what the test is named for by checking the backend and its connection pool are the same objects across two searches. --- .../strands/test_web_search_integration.py | 22 +++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/tests_integ/tools/integrations/strands/test_web_search_integration.py b/tests_integ/tools/integrations/strands/test_web_search_integration.py index 4c73007f..570e0f01 100644 --- a/tests_integ/tools/integrations/strands/test_web_search_integration.py +++ b/tests_integ/tools/integrations/strands/test_web_search_integration.py @@ -30,6 +30,11 @@ DEFAULT_MODEL_ID = "us.anthropic.claude-sonnet-4-5-20250929-v1:0" +def _numbered_blocks(text): + """Count the results in formatted output, by the lines that open a block.""" + return sum(1 for line in text.splitlines() if line[:1].isdigit() and line[1:3] == ". ") + + @pytest.mark.integration class TestAgentCoreWebSearchIntegration: """AgentCoreWebSearch against a live gateway target.""" @@ -73,7 +78,10 @@ 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 + # Counting the lines that open a numbered block, rather than searching for + # "3. " anywhere: extracts are arbitrary web prose and quite often contain a + # digit followed by a period themselves. + assert _numbered_blocks(text) == 2 def test_search_with_domain_filter(self): """Needs connector version 1.2.0 or later on the target. @@ -109,11 +117,21 @@ def test_tool_name_discovery_without_a_target_name(self): assert tool._client.backend._tool_name.endswith("WebSearch") def test_the_client_is_reused_across_searches(self): + """Two searches share one backend and one connection pool. + + Not asserted through ``Mcp-Session-Id``: that header is optional in MCP and + this connector does not send one, so the backend's session id stays None even + though the transport underneath is being reused. + """ with self._search_tool() as tool: + backend = tool._client.backend + pool = backend._http + self._search(tool, "first query", max_results=1) self._search(tool, "second query", max_results=1) - assert tool._client.backend._mcp_session_id + assert tool._client.backend is backend + assert tool._client.backend._http is pool 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.