diff --git a/litellm/llms/base_llm/search/transformation.py b/litellm/llms/base_llm/search/transformation.py index dee67e0b100f..7668c6132d6d 100644 --- a/litellm/llms/base_llm/search/transformation.py +++ b/litellm/llms/base_llm/search/transformation.py @@ -183,6 +183,29 @@ def validate_environment( """ return headers + def sign_request( + self, + headers: dict[str, str], # mutable-ok: matches the request header dict every other hook on this base takes + optional_params: dict[str, object], # mutable-ok: matches every other hook on this base + request_data: dict[str, object] | list[dict[str, object]], # mutable-ok: transform_search_request's body + api_base: str, + api_key: str | None = None, + ) -> tuple[dict[str, str], bytes | None]: # mutable-ok: the handler passes these headers straight to httpx + """ + OPTIONAL + + Sign the request. Providers like Bedrock AgentCore need to SigV4-sign + the request before sending it to the API. + + For all other providers, this is a no-op and we just return the headers. + + Returns: + Tuple of (headers, signed_json_body). When signed_json_body is not + None, the handler MUST send it verbatim as the request body — + re-serializing the payload would invalidate the signature. + """ + return headers, None + def get_complete_url( self, api_base: str | None, diff --git a/litellm/llms/bedrock/search/__init__.py b/litellm/llms/bedrock/search/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/litellm/llms/bedrock/search/transformation.py b/litellm/llms/bedrock/search/transformation.py new file mode 100644 index 000000000000..920e566c9dd2 --- /dev/null +++ b/litellm/llms/bedrock/search/transformation.py @@ -0,0 +1,455 @@ +""" +Calls an Amazon Bedrock AgentCore Gateway web-search target (MCP protocol) to search the web. + +Web Search on Amazon Bedrock AgentCore exposes Amazon's managed web index through +an AgentCore Gateway MCP endpoint. + +AWS docs: https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/gateway-target-connector-web-search-tool.html + +Authentication (matches the gateway's inbound authorizer type): +- AWS_IAM gateway: the request is SigV4-signed. Credentials come from explicit + params (aws_access_key_id / aws_secret_access_key / aws_session_token / + aws_region_name, also settable in a proxy search_tools entry) or the + standard AWS credential chain (env / profile / IRSA / assumed role) +- CUSTOM_JWT gateway: pass the OAuth2 bearer token (e.g. Cognito + client_credentials) as api_key, or set AGENTCORE_GATEWAY_TOKEN + +Setup: + 1. Create an AgentCore Gateway with a web-search connector target + 2. Set AGENTCORE_GATEWAY_URL (or pass api_base) to the gateway MCP endpoint, e.g. + https://.gateway.bedrock-agentcore..amazonaws.com/mcp + 3. AWS_IAM: ensure the credentials allow bedrock-agentcore:InvokeGateway + CUSTOM_JWT: set AGENTCORE_GATEWAY_TOKEN (or pass api_key) + +Usage: + response = litellm.search( + query="latest AI developments", + search_provider="agentcore", + max_results=5, + aws_access_key_id="...", # optional, omit to use the default chain + aws_secret_access_key="...", + ) +""" + +import json +import re +from collections.abc import Iterator, Mapping, Sequence +from typing import Final + +import httpx + +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj +from litellm.llms.base_llm.chat.transformation import BaseLLMException +from litellm.llms.base_llm.search.transformation import ( + BaseSearchConfig, + SearchResponse, + SearchResult, +) +from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM +from litellm.llms.bedrock.common_utils import BedrockError +from litellm.secret_managers.main import get_secret_str + +# AgentCore web-search rejects queries longer than 200 characters +AGENTCORE_MAX_QUERY_LENGTH: Final = 200 + +# The provider contract documents a default of 10 results, send it explicitly +# so the gateway can't silently apply a different default. +AGENTCORE_DEFAULT_MAX_RESULTS: Final = 10 + +# Default MCP tool name for a gateway web-search connector target: +# "___". Override with AGENTCORE_SEARCH_TOOL_NAME +# or optional_params["tool_name"] when the target uses a custom name. +AGENTCORE_DEFAULT_TOOL_NAME: Final = "web-search-tool___WebSearch" + +# All web-search connector tools share this suffix; rejecting other names keeps +# a caller-supplied tool_name from invoking unrelated tools on the same gateway +# with the proxy's credentials. +AGENTCORE_TOOL_NAME_SUFFIX: Final = "___WebSearch" + +# MCP revision this provider speaks. Sent on every request because the gateway is +# called statelessly, without an initialize handshake to negotiate a version. +# AgentCore gateways whose protocolConfiguration leaves supportedVersions unset +# accept only 2025-03-26 and reject anything newer with a -32600 error, so that +# is the default; a gateway pinned to another version needs +# AGENTCORE_MCP_PROTOCOL_VERSION set to match. +AGENTCORE_DEFAULT_MCP_PROTOCOL_VERSION: Final = "2025-03-26" + +# Matched against the URL host so a crafted path or query string can't pass for +# a gateway hostname. +_GATEWAY_HOST_PATTERN: Final = re.compile(r"[a-z0-9-]+\.gateway\.bedrock-agentcore\.([a-z0-9-]+)\.amazonaws\.com") + +_SSE_EVENT_SEPARATOR: Final = re.compile(r"\r?\n[ \t]*\r?\n") + +_SSE_LINE_PREFIXES: Final = ("event:", "data:", ":", "id:", "retry:") + + +def _gateway_host_match(api_base: str) -> re.Match[str] | None: + return _GATEWAY_HOST_PATTERN.fullmatch(httpx.URL(api_base).host) + + +_LOOPBACK_HOSTS: Final = frozenset({"localhost", "127.0.0.1", "::1"}) + + +def _credential_safe_transport(api_base: str) -> bool: + url: Final = httpx.URL(api_base) + return url.scheme == "https" or url.host in _LOOPBACK_HOSTS + + +def _string_field(item: Mapping[str, object], *keys: str) -> str | None: + return next( + (value for key in keys if isinstance(value := item.get(key), str) and value), + None, + ) + + +def _to_search_result(item: Mapping[str, object]) -> SearchResult: + return SearchResult( + title=_string_field(item, "title") or "", + url=_string_field(item, "url") or "", + snippet=_string_field(item, "text", "snippet") or "", + date=_string_field(item, "publishedDate", "date"), + last_updated=None, + ) + + +def _result_items(parsed: object) -> tuple[Mapping[str, object], ...]: + items: Final = parsed.get("results", ()) if isinstance(parsed, Mapping) else parsed + if not isinstance(items, Sequence) or isinstance(items, (str, bytes)): + return () + return tuple(item for item in items if isinstance(item, Mapping)) + + +def _parse_result_items(raw_text: object) -> tuple[Mapping[str, object], ...]: + """ + Parse one MCP text block into the search result objects it carries. + + A block holds either a JSON list of results or a {"results": [...]} object; + anything unparseable is skipped rather than failing the whole response. + """ + if not isinstance(raw_text, str): + return () + try: + parsed: Final = json.loads(raw_text) + except json.JSONDecodeError: + return () + return _result_items(parsed) + + +def _iter_sse_events(text: str) -> Iterator[Mapping[str, object]]: + """ + Yield the JSON payload of each SSE event in a Streamable HTTP MCP response. + + Per the SSE spec an event's data is the concatenation of all its ``data:`` + lines (joined with newlines), and a stream may carry several events, e.g. + progress notifications before the JSON-RPC response. + """ + for chunk in _SSE_EVENT_SEPARATOR.split(text): + payload = "\n".join(line[len("data:") :].lstrip() for line in chunk.splitlines() if line.startswith("data:")) + if not payload: + continue + try: + parsed = json.loads(payload) + except json.JSONDecodeError: + continue + if isinstance(parsed, dict): + yield parsed + + +class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM): + def __init__(self) -> None: + BaseSearchConfig.__init__(self) + BaseAWSLLM.__init__(self) + + @staticmethod + def ui_friendly_name() -> str: + return "Web Search on Amazon Bedrock" + + def validate_environment( + self, + headers: dict, # mutable-ok: BaseSearchConfig hands providers the mutable request header dict + api_key: str | None = None, + api_base: str | None = None, + **kwargs: object, # kwargs-ok: BaseSearchConfig.validate_environment forwards provider-specific extras + ) -> dict: # mutable-ok: the handler passes these headers straight to httpx, which wants a dict + """ + Set MCP transport headers. Per the MCP Streamable HTTP transport spec, + the client MUST accept both application/json and text/event-stream, and + declare its protocol revision with MCP-Protocol-Version. + + Authentication itself happens in sign_request(): bearer token for + CUSTOM_JWT gateways, AWS SigV4 for AWS_IAM gateways. + """ + return { # mutable-ok: httpx request headers are a dict + **headers, + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", + "MCP-Protocol-Version": get_secret_str("AGENTCORE_MCP_PROTOCOL_VERSION") + or AGENTCORE_DEFAULT_MCP_PROTOCOL_VERSION, + } + + def get_complete_url( + self, + api_base: str | None, + optional_params: dict, # mutable-ok: BaseSearchConfig passes optional params as a dict + data: dict | list[dict] | None = None, # mutable-ok: BaseSearchConfig request bodies are JSON dicts + **kwargs: object, # kwargs-ok: BaseSearchConfig.get_complete_url forwards provider-specific extras + ) -> str: + gateway_url: Final = api_base or get_secret_str("AGENTCORE_GATEWAY_URL") + if not gateway_url: + raise ValueError( + "AGENTCORE_GATEWAY_URL is not set. Set it to your AgentCore Gateway MCP " + "endpoint (https://.gateway.bedrock-agentcore." + ".amazonaws.com/mcp) or pass api_base." + ) + return gateway_url + + def transform_search_request( + self, + query: str | list[str], # mutable-ok: BaseSearchConfig accepts a list of queries + optional_params: dict, # mutable-ok: BaseSearchConfig passes optional params as a dict + **kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_request forwards provider-specific extras + ) -> dict: # mutable-ok: the JSON-RPC body is serialized as a JSON object + """ + Transform Search request to an MCP tools/call request. + + Args: + query: Search query (string or list of strings). AgentCore only + supports single string queries; lists are joined with spaces. + optional_params: Optional parameters for the request + - max_results: Maximum number of results (1-25), default 10 + - tool_name: Override the MCP tool name of the gateway target + + Returns: + Dict with the JSON-RPC 2.0 request body + """ + joined_query: Final = " ".join(query) if isinstance(query, list) else query + tool_name: Final = ( + optional_params.get("tool_name") + or get_secret_str("AGENTCORE_SEARCH_TOOL_NAME") + or AGENTCORE_DEFAULT_TOOL_NAME + ) + if not tool_name.endswith(AGENTCORE_TOOL_NAME_SUFFIX): + raise ValueError( + f"Invalid AgentCore search tool_name '{tool_name}': must end with " + f"'{AGENTCORE_TOOL_NAME_SUFFIX}' (a web-search connector tool). " + "Other gateway tools cannot be invoked through this provider." + ) + + return { # mutable-ok: JSON-RPC request bodies are JSON objects + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { # mutable-ok: JSON-RPC request bodies are JSON objects + "name": tool_name, + "arguments": { # mutable-ok: JSON-RPC request bodies are JSON objects + "query": joined_query[:AGENTCORE_MAX_QUERY_LENGTH], + "maxResults": optional_params.get("max_results", AGENTCORE_DEFAULT_MAX_RESULTS), + }, + }, + } + + def sign_request( + self, + headers: dict[str, str], # mutable-ok: BaseSearchConfig hands providers the mutable request header dict + optional_params: dict[str, object], # mutable-ok: BaseSearchConfig passes optional params as a dict + request_data: dict[str, object] | list[dict[str, object]], # mutable-ok: request bodies are JSON dicts + api_base: str, + api_key: str | None = None, + ) -> tuple[dict[str, str], bytes | None]: # mutable-ok: BaseSearchConfig.sign_request returns httpx headers + """ + Authenticate the MCP request. + + CUSTOM_JWT gateways: attach the caller's OAuth2 bearer token (api_key + or AGENTCORE_GATEWAY_TOKEN), no AWS credentials involved. + + AWS_IAM gateways: SigV4-sign with the bedrock-agentcore service name. + """ + if not isinstance(request_data, dict): + raise TypeError("AgentCore search expects a single dict request body") + + if not _credential_safe_transport(api_base): + raise ValueError( + f"Refusing to send AgentCore credentials over plaintext HTTP to '{api_base}': a bearer " + "token or SigV4 signature would be readable in transit. Use an https gateway URL " + "(plain http is allowed only for localhost)." + ) + + # Server-managed credentials only go to a trusted host, otherwise an + # authenticated caller could point api_base at their own server (e.g. via + # /search_tools/test_connection) and collect AGENTCORE_GATEWAY_TOKEN or a + # SigV4 signature with the proxy's credential scope and session token. + gateway_host_match: Final = _gateway_host_match(api_base) + bearer_token: Final = self.resolve_server_api_key( + caller_api_key=api_key, + caller_api_base=api_base, + key_env_vars=("AGENTCORE_GATEWAY_TOKEN",), + base_env_var="AGENTCORE_GATEWAY_URL", + default_api_base=api_base if gateway_host_match else None, + ) + if bearer_token: + bearer_headers: Final = { # mutable-ok: httpx request headers are a dict + **headers, + "Authorization": f"Bearer {bearer_token}", + } + return bearer_headers, json.dumps(request_data).encode() + + if gateway_host_match is None and not self._is_configured_gateway(api_base): + raise ValueError( + f"Refusing to send SigV4-signed AgentCore requests to '{api_base}': it is neither an " + "AgentCore gateway hostname nor the host in AGENTCORE_GATEWAY_URL. Set " + "AGENTCORE_GATEWAY_URL to authorize a custom gateway hostname." + ) + + signing_params: Final = ( + optional_params + if optional_params.get("aws_region_name") is not None + else { # mutable-ok: BaseAWSLLM._sign_request takes optional params as a dict + **optional_params, + "aws_region_name": self._signing_region(api_base), + } + ) + + # api_key="" (not None, but falsy) disables BaseAWSLLM's fallback to the + # AWS_BEARER_TOKEN_BEDROCK env var: that token is a Bedrock Runtime + # credential and must not be sent to an AgentCore gateway. + return self._sign_request( + service_name="bedrock-agentcore", + headers=headers, + optional_params=signing_params, + request_data=request_data, + api_base=api_base, + api_key="", + ) + + @staticmethod + def _is_configured_gateway(api_base: str) -> bool: + configured: Final = get_secret_str("AGENTCORE_GATEWAY_URL") + if not configured: + return False + return httpx.URL(configured).host == httpx.URL(api_base).host + + @staticmethod + def _signing_region(api_base: str) -> str: + """ + Resolve the SigV4 signing region, which must match the gateway's region. + + Standard gateway hostnames carry it, so callers don't have to set + aws_region_name to a region different from their default. For custom or + private hostnames, defer to the AWS configuration chain (env vars and + the shared config / profile region), and error out when that yields + nothing rather than silently signing for a guessed region the gateway + would reject with a confusing auth error. + """ + match: Final = _gateway_host_match(api_base) + if match: + return match.group(1) + + # boto3's session resolution covers env vars AND the AWS shared config + # (profile region), unlike BaseAWSLLM's helper, which silently defaults + # to us-west-2 when nothing is configured. + import boto3 + + configured_region: Final = boto3.Session().region_name + if configured_region: + return configured_region + raise ValueError( + f"Cannot derive the SigV4 signing region from api_base '{api_base}' " + "or the AWS configuration chain. Set aws_region_name (or AWS_DEFAULT_REGION / " + "a profile region) to the gateway's region when using a custom hostname." + ) + + def transform_search_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + **kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_response forwards provider-specific extras + ) -> SearchResponse: + """ + Transform an MCP tools/call response to LiteLLM unified SearchResponse. + + The gateway returns JSON-RPC (as plain JSON or a single-message SSE + stream) whose result.content[] text blocks contain a JSON list of + {title, url, date/publishedDate, text} entries. Web-search connector + 1.1.0 and later repeat that list in result.structuredContent, which is + the only machine-readable copy when the text block holds prose instead. + """ + response_json: Final = self._parse_mcp_body(raw_response) + + error: Final = response_json.get("error") + if error is not None: + raise BedrockError( + status_code=raw_response.status_code if raw_response.status_code >= 400 else 502, + message=f"AgentCore gateway MCP error: {error}", + ) + + # A failed tools/call is reported in-band, as HTTP 200 with result.isError + # and the failure text where the results would be. + result: Final = response_json.get("result") + if isinstance(result, dict) and result.get("isError"): + raise BedrockError( + status_code=raw_response.status_code if raw_response.status_code >= 400 else 502, + message=f"AgentCore web search tool error: {self._tool_error_message(response_json)}", + ) + + text_items: Final = tuple( + item for block in self._text_blocks(response_json) for item in _parse_result_items(block.get("text")) + ) + structured: Final = result.get("structuredContent") if isinstance(result, Mapping) else None + items: Final = text_items or _result_items(structured) + + results: Final = [_to_search_result(item) for item in items] # mutable-ok: pydantic list field + + return SearchResponse(results=results, object="search") + + def _tool_error_message(self, response_json: Mapping[str, object]) -> str: + texts: Final = tuple( + text for block in self._text_blocks(response_json) if isinstance(text := block.get("text"), str) + ) + return " ".join(texts) if texts else json.dumps(response_json.get("result"))[:500] + + @staticmethod + def _text_blocks(response_json: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: + result: Final = response_json.get("result") + content: Final = result.get("content") if isinstance(result, dict) else None + if not isinstance(content, Sequence) or isinstance(content, (str, bytes)): + return () + return tuple(block for block in content if isinstance(block, dict) and block.get("type") == "text") + + @staticmethod + def _parse_mcp_body(raw_response: httpx.Response) -> Mapping[str, object]: + """ + Parse a JSON or SSE-framed (Streamable HTTP transport) MCP response. + + Return the event whose payload carries the JSON-RPC response, i.e. one + containing ``result`` or ``error``, falling back to the last event when + the stream carries only notifications. + """ + text: Final = raw_response.text + if not text.lstrip().startswith(_SSE_LINE_PREFIXES): + return raw_response.json() + + events: Final = tuple(_iter_sse_events(text)) + response_event: Final = next( + (event for event in events if "result" in event or "error" in event), + None, + ) + if response_event is not None: + return response_event + if events: + return events[-1] + raise BedrockError( + status_code=502, + message=f"AgentCore gateway returned SSE without a JSON data frame: {text[:200]}", + ) + + def get_error_class( + self, + error_message: str, + status_code: int, + headers: dict, # mutable-ok: BaseSearchConfig.get_error_class takes the response headers as a dict + ) -> Exception: + return BaseLLMException( + status_code=status_code, + message=error_message, + headers=headers, + ) diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index d5b47466477f..dccd895ce3a6 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -1763,6 +1763,14 @@ def search( api_key=api_key, ) + signed_headers, signed_json_body = provider_config.sign_request( + headers=headers, + optional_params=optional_params, + request_data=data, + api_base=complete_url, + api_key=api_key, + ) + ## LOGGING logging_obj.pre_call( input=query if isinstance(query, str) else str(query), @@ -1786,14 +1794,15 @@ def search( # Note: timeout is set on the client itself, not per-request for GET response = client.get( url=complete_url, - headers=headers, + headers=signed_headers, ) else: - # Make POST request with JSON data + # A signed body must be sent verbatim, re-serializing it would break the signature response = client.post( url=complete_url, - headers=headers, - json=data, + headers=signed_headers, + data=signed_json_body, + json=data if signed_json_body is None else None, timeout=timeout, ) except Exception as e: @@ -1847,6 +1856,14 @@ async def async_search( api_key=api_key, ) + signed_headers, signed_json_body = provider_config.sign_request( + headers=headers, + optional_params=optional_params, + request_data=data, + api_base=complete_url, + api_key=api_key, + ) + ## LOGGING logging_obj.pre_call( input=query if isinstance(query, str) else str(query), @@ -1875,14 +1892,15 @@ async def async_search( # Note: timeout is set on the client itself, not per-request for GET response = await async_httpx_client.get( url=complete_url, - headers=headers, + headers=signed_headers, ) else: - # Make async POST request with JSON data + # A signed body must be sent verbatim, re-serializing it would break the signature response = await async_httpx_client.post( url=complete_url, - headers=headers, - json=data, + headers=signed_headers, + data=signed_json_body, + json=data if signed_json_body is None else None, timeout=timeout, ) except Exception as e: diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 78b53cefc53d..066d6859a32e 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -16435,6 +16435,14 @@ "notes": "APISerpent deep search (/api/search), multi-engine (Google, Bing, Yahoo, DuckDuckGo). Pricing: $0.60/1k searches." } }, + "agentcore/search": { + "input_cost_per_query": 0.0, + "litellm_provider": "agentcore", + "mode": "search", + "metadata": { + "notes": "Web Search on Amazon Bedrock AgentCore, billed by AWS on the gateway" + } + }, "tinyfish/search": { "input_cost_per_query": 0.0, "litellm_provider": "tinyfish", diff --git a/litellm/proxy/example_config_yaml/agentcore_websearch_config.yaml b/litellm/proxy/example_config_yaml/agentcore_websearch_config.yaml new file mode 100644 index 000000000000..12402095c4d7 --- /dev/null +++ b/litellm/proxy/example_config_yaml/agentcore_websearch_config.yaml @@ -0,0 +1,40 @@ +# Claude Code / Anthropic-native web search on Bedrock, backed by +# Amazon Bedrock AgentCore Web Search (AWS-managed web index, no third-party +# search API). See litellm/llms/bedrock/search/transformation.py for details. + +model_list: + - model_name: claude-sonnet + litellm_params: + model: bedrock/us.anthropic.claude-sonnet-5 + aws_region_name: us-east-1 + +search_tools: + - search_tool_name: agentcore-search + litellm_params: + search_provider: agentcore + # Your AgentCore Gateway MCP endpoint (gateway must have a `web-search` + # connector target). Alternatively set the AGENTCORE_GATEWAY_URL env var. + api_base: https://.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp + + # The gateway exposes the connector as "___WebSearch". + # Default is "web-search-tool___WebSearch", matching the target name used + # in the AWS docs' boto3/CLI setup examples. If your target was created + # with a different name (misconfiguration surfaces as an MCP "tool not + # found" error), set the AGENTCORE_SEARCH_TOOL_NAME env var or pass + # tool_name in the request body. The search router forwards only + # search_provider / api_key / api_base from this litellm_params block, + # so a tool_name set here would be silently ignored. + + # AWS_IAM gateway (default): SigV4-signed using the standard AWS + # credential chain (env / profile / IRSA / instance role). Explicit + # aws_access_key_id / aws_secret_access_key set here would be silently + # ignored for the same reason; pass them per request instead. + + # CUSTOM_JWT gateway alternative — OAuth2 bearer token instead of SigV4: + # api_key: os.environ/AGENTCORE_GATEWAY_TOKEN + +litellm_settings: + callbacks: ["websearch_interception"] + websearch_interception_params: + enabled_providers: ["bedrock"] + search_tool_name: agentcore-search diff --git a/litellm/types/utils.py b/litellm/types/utils.py index ae9395fc851f..f0502e52703f 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3791,6 +3791,7 @@ class SearchProviders(str, Enum): YOU_COM = "you_com" APISERPENT = "apiserpent" TINYFISH = "tinyfish" + AGENTCORE = "agentcore" NIMBLE = "nimble" diff --git a/litellm/utils.py b/litellm/utils.py index 73099eb47f46..f4f84ffe3e1f 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -9062,6 +9062,7 @@ def get_provider_search_config( from litellm.llms.apiserpent.search.transformation import ( APISerpentSearchConfig, ) + from litellm.llms.bedrock.search.transformation import AgentCoreSearchConfig from litellm.llms.brave.search.transformation import BraveSearchConfig from litellm.llms.dataforseo.search.transformation import DataForSEOSearchConfig from litellm.llms.duckduckgo.search.transformation import DuckDuckGoSearchConfig @@ -9100,6 +9101,7 @@ def get_provider_search_config( SearchProviders.YOU_COM: YouComSearchConfig, SearchProviders.APISERPENT: APISerpentSearchConfig, SearchProviders.TINYFISH: TinyfishSearchConfig, + SearchProviders.AGENTCORE: AgentCoreSearchConfig, SearchProviders.NIMBLE: NimbleSearchConfig, } config_class: Final = PROVIDER_TO_CONFIG_MAP.get(provider, None) diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index 78b53cefc53d..066d6859a32e 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -16435,6 +16435,14 @@ "notes": "APISerpent deep search (/api/search), multi-engine (Google, Bing, Yahoo, DuckDuckGo). Pricing: $0.60/1k searches." } }, + "agentcore/search": { + "input_cost_per_query": 0.0, + "litellm_provider": "agentcore", + "mode": "search", + "metadata": { + "notes": "Web Search on Amazon Bedrock AgentCore, billed by AWS on the gateway" + } + }, "tinyfish/search": { "input_cost_per_query": 0.0, "litellm_provider": "tinyfish", diff --git a/tests/test_litellm/llms/bedrock/search/test_agentcore_search_transformation.py b/tests/test_litellm/llms/bedrock/search/test_agentcore_search_transformation.py new file mode 100644 index 000000000000..950336c7ad03 --- /dev/null +++ b/tests/test_litellm/llms/bedrock/search/test_agentcore_search_transformation.py @@ -0,0 +1,637 @@ +""" +Tests for Amazon Bedrock AgentCore Web Search integration. + +Mirror of tests/search_tests/test_agentcore_search.py placed in the +test_litellm tree so the AgentCoreSearchConfig transformation is exercised by +the sharded CI (coverage collection runs against this tree). +""" + +import json +import os + +import pytest +from unittest.mock import AsyncMock, patch, MagicMock + +import litellm +from litellm.llms.bedrock.search.transformation import ( + AGENTCORE_DEFAULT_MCP_PROTOCOL_VERSION, + AgentCoreSearchConfig, +) + +GATEWAY_URL = "https://testgateway-abc123.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp" + +MCP_RESULTS = [ + { + "title": "Test Result 1", + "url": "https://example.com/1", + "text": "Snippet for result 1", + "publishedDate": "2026-06-16", + }, + { + "title": "Test Result 2", + "url": "https://example.com/2", + "text": "Snippet for result 2", + }, +] + + +def _mcp_response_body() -> dict: + return { + "jsonrpc": "2.0", + "id": 1, + "result": {"content": [{"type": "text", "text": json.dumps(MCP_RESULTS)}]}, + } + + +def _make_mock_response(json_body: dict = None, text: str = None) -> MagicMock: + mock_response = MagicMock() + mock_response.status_code = 200 + if text is not None: + mock_response.text = text + else: + mock_response.text = json.dumps(json_body) + mock_response.json.return_value = json_body + return mock_response + + +class TestAgentCoreSearch: + """ + Tests for AgentCore Web Search functionality with mocked network/signing. + """ + + @pytest.mark.asyncio + async def test_agentcore_search_request_payload(self): + """Validates the MCP tools/call payload and SigV4 signing without real AWS calls.""" + os.environ["AGENTCORE_GATEWAY_URL"] = GATEWAY_URL + + mock_response = _make_mock_response(_mcp_response_body()) + + with ( + patch( + "litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post", + new_callable=AsyncMock, + ) as mock_post, + patch.object( + AgentCoreSearchConfig, + "_sign_request", + return_value=( + {"Authorization": "AWS4-HMAC-SHA256 test", "Content-Type": "application/json"}, + json.dumps({"signed": True}).encode(), + ), + ) as mock_sign, + ): + mock_post.return_value = mock_response + + response = await litellm.asearch( + query="latest developments in AI", + search_provider="agentcore", + max_results=5, + ) + + mock_post.assert_called_once() + call_kwargs = mock_post.call_args.kwargs + assert call_kwargs["url"] == GATEWAY_URL + # Signed body must be sent verbatim + assert call_kwargs["data"] == json.dumps({"signed": True}).encode() + assert call_kwargs["json"] is None + + # Signing was invoked with the MCP request + mock_sign.assert_called_once() + sign_kwargs = mock_sign.call_args.kwargs + request_data = sign_kwargs["request_data"] + assert request_data["method"] == "tools/call" + assert request_data["params"]["name"] == "web-search-tool___WebSearch" + assert request_data["params"]["arguments"]["query"] == "latest developments in AI" + assert request_data["params"]["arguments"]["maxResults"] == 5 + assert sign_kwargs["service_name"] == "bedrock-agentcore" + + assert len(response.results) == 2 + assert response.results[0].title == "Test Result 1" + assert response.results[0].url == "https://example.com/1" + assert response.results[0].snippet == "Snippet for result 1" + assert response.results[0].date == "2026-06-16" + + def test_transform_search_request_query_truncation(self): + """AgentCore rejects queries > 200 chars; the request must truncate.""" + config = AgentCoreSearchConfig() + long_query = "a" * 300 + data = config.transform_search_request(query=long_query, optional_params={}) + assert len(data["params"]["arguments"]["query"]) == 200 + + def test_transform_search_request_joins_list_queries(self): + config = AgentCoreSearchConfig() + data = config.transform_search_request(query=["foo", "bar"], optional_params={}) + assert data["params"]["arguments"]["query"] == "foo bar" + + def test_transform_search_request_custom_tool_name(self): + config = AgentCoreSearchConfig() + data = config.transform_search_request(query="q", optional_params={"tool_name": "my-target___WebSearch"}) + assert data["params"]["name"] == "my-target___WebSearch" + + def test_transform_search_request_rejects_non_websearch_tool_name(self): + """A caller-supplied tool_name must not reach other tools on the gateway.""" + config = AgentCoreSearchConfig() + with pytest.raises(ValueError, match="must end with"): + config.transform_search_request(query="q", optional_params={"tool_name": "admin-target___DeleteUser"}) + + def test_transform_search_request_sends_documented_default_max_results(self): + """The documented default of 10 is sent explicitly, not left to the gateway.""" + config = AgentCoreSearchConfig() + data = config.transform_search_request(query="q", optional_params={}) + assert data["params"]["arguments"]["maxResults"] == 10 + + def test_get_complete_url_requires_gateway_url(self): + config = AgentCoreSearchConfig() + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + with pytest.raises(ValueError, match="AGENTCORE_GATEWAY_URL"): + config.get_complete_url(api_base=None, optional_params={}) + + def test_get_complete_url_prefers_api_base(self): + config = AgentCoreSearchConfig() + assert config.get_complete_url(api_base=GATEWAY_URL, optional_params={}) == GATEWAY_URL + + def test_validate_environment_sets_mcp_headers(self): + """MCP Streamable HTTP requires accepting both JSON and SSE, and declaring + the protocol revision the client speaks.""" + config = AgentCoreSearchConfig() + headers = config.validate_environment(headers={}) + assert headers["Accept"] == "application/json, text/event-stream" + assert headers["Content-Type"] == "application/json" + assert headers["MCP-Protocol-Version"] == AGENTCORE_DEFAULT_MCP_PROTOCOL_VERSION + + def test_default_protocol_version_is_the_agentcore_gateway_default(self): + """A default AgentCore gateway supports only 2025-03-26 and answers + -32600 to anything newer, so that exact revision must be the default.""" + assert AGENTCORE_DEFAULT_MCP_PROTOCOL_VERSION == "2025-03-26" + + def test_protocol_version_env_override_wins(self): + """A gateway pinned to a newer supportedVersions list needs the header + to match, so AGENTCORE_MCP_PROTOCOL_VERSION must override the default.""" + config = AgentCoreSearchConfig() + with patch.dict(os.environ, {"AGENTCORE_MCP_PROTOCOL_VERSION": "2025-06-18"}): + headers = config.validate_environment(headers={}) + assert headers["MCP-Protocol-Version"] == "2025-06-18" + + def test_protocol_version_header_survives_signing(self): + """Both auth paths must keep the MCP-Protocol-Version header on the wire.""" + config = AgentCoreSearchConfig() + headers = config.validate_environment(headers={}) + + bearer_headers, _ = config.sign_request( + headers=headers, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=GATEWAY_URL, + api_key="test-jwt-token", + ) + assert bearer_headers["MCP-Protocol-Version"] == AGENTCORE_DEFAULT_MCP_PROTOCOL_VERSION + + with patch.dict( + os.environ, + { + "AWS_ACCESS_KEY_ID": "AKIAIOSFODNN7EXAMPLE", + "AWS_SECRET_ACCESS_KEY": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + }, + ): + signed_headers, _ = config.sign_request( + headers=headers, + optional_params={"aws_region_name": "us-east-1"}, + request_data={"jsonrpc": "2.0"}, + api_base=GATEWAY_URL, + ) + assert signed_headers["Authorization"].startswith("AWS4-HMAC-SHA256") + assert signed_headers["MCP-Protocol-Version"] == AGENTCORE_DEFAULT_MCP_PROTOCOL_VERSION + + def test_transform_search_response_parses_sse_frame(self): + """Gateway may answer with an SSE-framed JSON-RPC message.""" + config = AgentCoreSearchConfig() + body = _mcp_response_body() + sse_text = f"event: message\ndata: {json.dumps(body)}\n\n" + mock_response = _make_mock_response(text=sse_text) + + response = config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + assert len(response.results) == 2 + assert response.results[1].url == "https://example.com/2" + + def test_transform_search_response_parses_multiline_sse_data(self): + """SSE data may be split across several data: lines (joined per spec).""" + config = AgentCoreSearchConfig() + pretty = json.dumps(_mcp_response_body(), indent=2) + sse_text = "event: message\n" + "\n".join(f"data: {line}" for line in pretty.splitlines()) + "\n\n" + mock_response = _make_mock_response(text=sse_text) + + response = config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + assert len(response.results) == 2 + + def test_transform_search_response_skips_progress_events(self): + """A progress notification before the JSON-RPC result must not shadow it.""" + config = AgentCoreSearchConfig() + progress = {"jsonrpc": "2.0", "method": "notifications/progress", "params": {"progress": 1}} + sse_text = ( + f"event: message\ndata: {json.dumps(progress)}\n\n" + f"event: message\ndata: {json.dumps(_mcp_response_body())}\n\n" + ) + mock_response = _make_mock_response(text=sse_text) + + response = config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + assert len(response.results) == 2 + + def test_transform_search_response_raises_on_mcp_error(self): + config = AgentCoreSearchConfig() + mock_response = _make_mock_response( + {"jsonrpc": "2.0", "id": 1, "error": {"code": -32601, "message": "tool not found"}} + ) + with pytest.raises(Exception, match="tool not found"): + config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + + def test_transform_search_response_raises_on_tool_error(self): + """A failed tools/call comes back as HTTP 200 with result.isError; it must not be + reported to the caller as a successful search with zero results.""" + config = AgentCoreSearchConfig() + mock_response = _make_mock_response( + { + "jsonrpc": "2.0", + "id": 1, + "result": { + "isError": True, + "content": [{"type": "text", "text": "AccessDeniedException: not authorized"}], + }, + } + ) + with pytest.raises(Exception, match="AccessDeniedException"): + config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + + def test_transform_search_response_reads_structured_content(self): + """Connector 1.1.0+ puts the machine-readable results in structuredContent and may + leave the text block as prose, which must not come back as an empty result list.""" + config = AgentCoreSearchConfig() + mock_response = _make_mock_response( + { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{"type": "text", "text": "Here is a prose summary of what I found."}], + "structuredContent": {"id": "824f89d0", "results": MCP_RESULTS}, + }, + } + ) + + response = config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + assert [result.title for result in response.results] == ["Test Result 1", "Test Result 2"] + assert response.results[0].url == "https://example.com/1" + assert response.results[0].snippet == "Snippet for result 1" + assert response.results[0].date == "2026-06-16" + + def test_transform_search_response_does_not_duplicate_structured_content(self): + """1.1.0+ repeats the same results in both places, so parsing both would double them.""" + config = AgentCoreSearchConfig() + body = _mcp_response_body() + body["result"]["structuredContent"] = {"results": MCP_RESULTS} + mock_response = _make_mock_response(body) + + response = config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + assert len(response.results) == 2 + + def test_transform_search_response_parses_crlf_framed_sse(self): + """SSE streams may be CRLF framed; events must still split into separate events.""" + config = AgentCoreSearchConfig() + progress = {"jsonrpc": "2.0", "method": "notifications/progress", "params": {"progress": 1}} + sse_text = ( + f"event: message\r\ndata: {json.dumps(progress)}\r\n\r\n" + f"event: message\r\ndata: {json.dumps(_mcp_response_body())}\r\n\r\n" + ) + mock_response = _make_mock_response(text=sse_text) + + response = config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + assert len(response.results) == 2 + assert response.results[0].title == "Test Result 1" + + def test_sign_request_uses_bearer_token_when_api_key_set(self): + """CUSTOM_JWT gateways: api_key is sent as a bearer token, no SigV4.""" + config = AgentCoreSearchConfig() + request_data = {"jsonrpc": "2.0", "id": 1} + + headers, signed_body = config.sign_request( + headers={"Content-Type": "application/json"}, + optional_params={}, + request_data=request_data, + api_base=GATEWAY_URL, + api_key="test-jwt-token", + ) + assert headers["Authorization"] == "Bearer test-jwt-token" + assert signed_body == json.dumps(request_data).encode() + + def test_sign_request_uses_bearer_token_from_env(self): + """Server token is attached when the request targets the configured gateway host.""" + config = AgentCoreSearchConfig() + os.environ["AGENTCORE_GATEWAY_TOKEN"] = "env-jwt-token" + os.environ["AGENTCORE_GATEWAY_URL"] = GATEWAY_URL + try: + headers, _ = config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=GATEWAY_URL, + ) + assert headers["Authorization"] == "Bearer env-jwt-token" + finally: + os.environ.pop("AGENTCORE_GATEWAY_TOKEN", None) + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + + def test_sign_request_refuses_server_token_to_untrusted_host(self): + """Server-managed token must not be sent to a caller-chosen api_base.""" + config = AgentCoreSearchConfig() + os.environ["AGENTCORE_GATEWAY_TOKEN"] = "env-jwt-token" + os.environ["AGENTCORE_GATEWAY_URL"] = GATEWAY_URL + try: + with pytest.raises(ValueError, match="Refusing to send"): + config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base="https://attacker.example.com/mcp", + ) + finally: + os.environ.pop("AGENTCORE_GATEWAY_TOKEN", None) + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + + def test_sign_request_uses_env_token_for_gateway_api_base_without_gateway_url(self): + """api_base pointing at a real gateway is a trusted destination for the env token, + so operators configuring api_base in yaml don't also need AGENTCORE_GATEWAY_URL.""" + config = AgentCoreSearchConfig() + os.environ["AGENTCORE_GATEWAY_TOKEN"] = "env-jwt-token" + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + try: + headers, _ = config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=GATEWAY_URL, + ) + assert headers["Authorization"] == "Bearer env-jwt-token" + finally: + os.environ.pop("AGENTCORE_GATEWAY_TOKEN", None) + + @pytest.mark.parametrize( + "untrusted_api_base", + [ + "https://attacker.example.com/mcp", + # gateway hostname in the path/query must not pass for the host + "https://attacker.example.com/gw.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp", + ], + ) + def test_sign_request_refuses_sigv4_to_untrusted_host(self, untrusted_api_base): + """A SigV4 signature carries the proxy's credential scope and session token, so it + must never be sent to a host that is not the operator's gateway.""" + config = AgentCoreSearchConfig() + os.environ.pop("AGENTCORE_GATEWAY_TOKEN", None) + os.environ["AGENTCORE_GATEWAY_URL"] = GATEWAY_URL + try: + with patch.object( + AgentCoreSearchConfig.__mro__[2], # BaseAWSLLM + "_sign_request", + return_value=({}, b"{}"), + ) as mock_base_sign: + with pytest.raises(ValueError, match="Refusing to send"): + config.sign_request( + headers={}, + optional_params={"aws_region_name": "us-east-1"}, + request_data={"jsonrpc": "2.0"}, + api_base=untrusted_api_base, + ) + mock_base_sign.assert_not_called() + finally: + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + + @pytest.mark.parametrize( + "plaintext_api_base", + [ + "http://gw.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp", + "http://internal-gateway.corp/mcp", + ], + ) + def test_sign_request_refuses_server_token_over_plaintext_http(self, plaintext_api_base): + """A trusted hostname over plain http would expose the bearer token to + network observers, so credentials only ride https (or localhost).""" + config = AgentCoreSearchConfig() + os.environ["AGENTCORE_GATEWAY_TOKEN"] = "env-jwt-token" + os.environ["AGENTCORE_GATEWAY_URL"] = plaintext_api_base + try: + with pytest.raises(ValueError, match="plaintext"): + config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=plaintext_api_base, + ) + finally: + os.environ.pop("AGENTCORE_GATEWAY_TOKEN", None) + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + + def test_sign_request_refuses_sigv4_over_plaintext_http(self): + """Same for SigV4: a signature over plain http is replayable by observers.""" + config = AgentCoreSearchConfig() + os.environ.pop("AGENTCORE_GATEWAY_TOKEN", None) + with patch.object( + AgentCoreSearchConfig.__mro__[2], # BaseAWSLLM + "_sign_request", + return_value=({}, b"{}"), + ) as mock_base_sign: + with pytest.raises(ValueError, match="plaintext"): + config.sign_request( + headers={}, + optional_params={"aws_region_name": "us-east-1"}, + request_data={"jsonrpc": "2.0"}, + api_base="http://gw.gateway.bedrock-agentcore.us-east-1.amazonaws.com/mcp", + ) + mock_base_sign.assert_not_called() + + def test_sign_request_allows_plain_http_for_localhost(self): + """Local development against an MCP stub on 127.0.0.1 keeps working.""" + config = AgentCoreSearchConfig() + os.environ["AGENTCORE_GATEWAY_TOKEN"] = "env-jwt-token" + os.environ["AGENTCORE_GATEWAY_URL"] = "http://127.0.0.1:8931/mcp" + try: + headers, _ = config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base="http://127.0.0.1:8931/mcp", + ) + assert headers["Authorization"] == "Bearer env-jwt-token" + finally: + os.environ.pop("AGENTCORE_GATEWAY_TOKEN", None) + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + + def test_sign_request_does_not_leak_bedrock_bearer_token(self): + """AWS_BEARER_TOKEN_BEDROCK is a Bedrock Runtime credential — it must not + replace SigV4 on requests to an AgentCore gateway.""" + config = AgentCoreSearchConfig() + + with patch.object( + AgentCoreSearchConfig.__mro__[2], # BaseAWSLLM + "_sign_request", + return_value=({}, b"{}"), + ) as mock_base_sign: + config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=GATEWAY_URL, + ) + # api_key="" (falsy, not None) disables the base class's + # AWS_BEARER_TOKEN_BEDROCK env fallback. + assert mock_base_sign.call_args.kwargs["api_key"] == "" + + def test_sign_request_custom_hostname_requires_region(self): + """Custom hostname + empty AWS config chain → clear error, no guessed region.""" + config = AgentCoreSearchConfig() + custom_url = "https://gateway.internal.example.com/mcp" + os.environ["AGENTCORE_GATEWAY_URL"] = custom_url + + mock_session = MagicMock() + mock_session.region_name = None # nothing configured anywhere + try: + with patch("boto3.Session", return_value=mock_session): + with pytest.raises(ValueError, match="signing region"): + config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=custom_url, + ) + finally: + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + + def test_sign_request_custom_hostname_uses_shared_config_region(self): + """Custom hostname + region from AWS shared config (profile) must be honored.""" + config = AgentCoreSearchConfig() + custom_url = "https://gateway.internal.example.com/mcp" + os.environ["AGENTCORE_GATEWAY_URL"] = custom_url + + mock_session = MagicMock() + mock_session.region_name = "eu-west-1" # e.g. from ~/.aws/config profile + try: + with ( + patch("boto3.Session", return_value=mock_session), + patch.object( + AgentCoreSearchConfig.__mro__[2], # BaseAWSLLM + "_sign_request", + return_value=({}, b"{}"), + ) as mock_base_sign, + ): + config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=custom_url, + ) + assert mock_base_sign.call_args.kwargs["optional_params"]["aws_region_name"] == "eu-west-1" + finally: + os.environ.pop("AGENTCORE_GATEWAY_URL", None) + + def test_sign_request_passes_explicit_aws_credentials(self): + """Explicit aws_* params (e.g. from a proxy search_tools entry) reach the signer.""" + config = AgentCoreSearchConfig() + + with patch.object( + AgentCoreSearchConfig.__mro__[2], # BaseAWSLLM + "_sign_request", + return_value=({}, b"{}"), + ) as mock_base_sign: + config.sign_request( + headers={}, + optional_params={ + "aws_access_key_id": "AKIATEST", + "aws_secret_access_key": "secret", + "aws_session_token": "token", + }, + request_data={"jsonrpc": "2.0"}, + api_base=GATEWAY_URL, + ) + passed = mock_base_sign.call_args.kwargs["optional_params"] + assert passed["aws_access_key_id"] == "AKIATEST" + assert passed["aws_secret_access_key"] == "secret" + assert passed["aws_session_token"] == "token" + + def test_sign_request_derives_region_from_gateway_url(self): + """Signing region must come from the gateway URL, not the caller's default region.""" + config = AgentCoreSearchConfig() + eu_url = "https://gw-x.gateway.bedrock-agentcore.eu-central-1.amazonaws.com/mcp" + + with patch.object( + AgentCoreSearchConfig.__mro__[2], # BaseAWSLLM + "_sign_request", + return_value=({}, b"{}"), + ) as mock_base_sign: + config.sign_request( + headers={}, + optional_params={}, + request_data={"jsonrpc": "2.0"}, + api_base=eu_url, + ) + assert mock_base_sign.call_args.kwargs["optional_params"]["aws_region_name"] == "eu-central-1" + + +class TestAgentCoreSearchEdgeCases: + """Branch coverage for response parsing and error mapping.""" + + def test_transform_search_response_skips_non_text_and_bad_json_blocks(self): + """Non-text blocks and unparseable text blocks are skipped, not fatal.""" + config = AgentCoreSearchConfig() + body = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [ + {"type": "image", "data": "..."}, + {"type": "text", "text": "not-json"}, + {"type": "text", "text": json.dumps(["scalar", {"title": "T", "url": "u", "text": "s"}])}, + ] + }, + } + mock_response = _make_mock_response(body) + + response = config.transform_search_response(raw_response=mock_response, logging_obj=MagicMock()) + # only the one dict item survives; non-dict list entries are skipped + assert len(response.results) == 1 + assert response.results[0].title == "T" + + def test_parse_mcp_body_sse_without_json_frame_raises(self): + """An SSE stream carrying no parseable JSON object is a 502.""" + config = AgentCoreSearchConfig() + mock_response = _make_mock_response(text="event: ping\ndata: not-json\n\n") + with pytest.raises(Exception, match="SSE without a JSON data frame"): + config._parse_mcp_body(mock_response) + + def test_parse_mcp_body_returns_last_event_when_no_result_frame(self): + """A stream of only notifications returns the last parsed event.""" + config = AgentCoreSearchConfig() + note = {"jsonrpc": "2.0", "method": "notifications/progress"} + mock_response = _make_mock_response(text=f"data: {json.dumps(note)}\n\n") + assert config._parse_mcp_body(mock_response) == note + + def test_sign_request_rejects_list_request_body(self): + config = AgentCoreSearchConfig() + with pytest.raises(TypeError, match="single dict"): + config.sign_request( + headers={}, + optional_params={}, + request_data=[{"jsonrpc": "2.0"}], + api_base=GATEWAY_URL, + ) + + def test_get_error_class_maps_status_and_message(self): + config = AgentCoreSearchConfig() + err = config.get_error_class(error_message="boom", status_code=503, headers={}) + assert getattr(err, "status_code", None) == 503 + assert "boom" in str(err) + + def test_search_cost_lookup_is_mapped(self, monkeypatch): + """Assert against the map in this checkout: the remote cost map litellm loads by + default only carries providers already released.""" + from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap + from litellm.search.cost_calculator import search_provider_cost_per_query + + monkeypatch.setattr(litellm, "model_cost", GetModelCostMap.load_local_model_cost_map()) + assert search_provider_cost_per_query(model="agentcore/search", custom_llm_provider="agentcore") == (0.0, 0.0)