Search for tools in your AgentCore gateway with a natural language query

If you enabled semantic search for your gateway when you created it, you can call the x_amz_bedrock_agentcore_search tool to search for tools in your gateway with a natural language query. Semantic search is particularly useful when you have many tools and need to find the most appropriate ones for your use case. To learn how to enable semantic search during gateway creation, see Creating your Gateway.

To search for a tool using this AgentCore tool, make the following POST request with the tools/call method to the gateway's MCP endpoint:

POST /mcp HTTP/1.1
Host: ${GatewayEndpoint}
Content-Type: application/json
Authorization: Bearer ${AccessToken}

{
  "jsonrpc": "2.0",
  "id": "${RequestName}",
  "method": "tools/call",
  "params": {
    "name": "x_amz_bedrock_agentcore_search",
    "arguments": {
      "query": ${Query}
    }
  }
}

Replace the following values:

• ${GatewayEndpoint} – The URL of the gateway, as provided in the response of the CreateGateway API.

• ${AccessToken} – The access token provided by the identity provider when you set up inbound authorization.

• ${RequestName} – A name for the request.

• ${Query} – A natural language query to search for tools.

The response returns a list of tools that are relevant to the query.

Code samples for tool searching

To see examples of using natural language queries to find tools in the gateway, select one of the following methods:

Python requests package

import requests
import json

def search_tools(gateway_url, access_token, query):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    payload = {
        "jsonrpc": "2.0",
        "id": "search-tools-request",
        "method": "tools/call",
        "params": {
            "name": "x_amz_bedrock_agentcore_search",
            "arguments": {
                "query": query
            }
        }
    }
    
    response = requests.post(gateway_url, headers=headers, json=payload)
    return response.json()

# Example usage
gateway_url = "https://${GatewayEndpoint}" # Replace with your actual gateway endpoint
access_token = "${AccessToken}" # Replace with your actual access token
results = search_tools(gateway_url, access_token, "find order information")
print(json.dumps(results, indent=2))

MCP Client

import json
from datetime import timedelta

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client


async def execute_mcp(
    url,
    headers=None
):
    headers = {**headers} if headers else {}
    async with streamablehttp_client(
       url=url,
       headers=headers,
    ) as (
        read_stream,
        write_stream,
        callA,
    ):
        async with ClientSession(read_stream, write_stream) as session:
            # 1. Perform initialization handshake
            print("Initializing MCP...")
            _init_response = await session.initialize()
            print(f"MCP Server Initialize successful! - {_init_response}")

            # 2. Invoke search tool
            print("Invoking search tool...")
            tool_name = "x_amz_bedrock_agentcore_search"
            args = {
                "query": "How do I process images?"
            }
            invoke_tool_response = await session.call_tool(
                tool_name,
                arguments=args,
                read_timeout_seconds=timedelta(seconds=60)
            )
            contents = invoke_tool_response.content
            for content in contents:
                text = content.text
                try:
                    content = json.dumps(json.loads(text), indent=4)
                except:
                    content = text
                print(
                    f"Invoke tool response: Name - {content}"
                )

Strands MCP Client

import asyncio
from strands import Agent
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
 
async def main():
    # Create strands agent
    agent = Agent()
    
    # Connect to MCP server via streamable HTTP
    async with streamablehttp_client("{gatewayURL}") as (read, write, get_session_id):
        async with ClientSession(read, write) as session:
            # Initialize MCP session
            await session.initialize()
            
            result = await session.call_tool('x_amz_bedrock_agentcore_searchaws___recommend', arguments={"query": "How do I process images?"})
            print(f"Result: {result.content}")
 
def create_streamable_http_transport(mcp_url: str, access_token: str):
       return streamablehttp_client(mcp_url, headers={"Authorization": f"Bearer {access_token}"})

if __name__ == "__main__":
    asyncio.run(main())

LangGraph MCP Client

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

url = ""
headers = {}

def filter_search_tool(
):
    mcp_client = MultiServerMCPClient(
        {
            "agent": {
                "transport": "streamable_http",
                "url": url,
                "headers": headers,
            }
        }
    )
    tools = asyncio.run(mcp_client.get_tools())
    builtin_search_tool = []
    for tool in tools:
        if tool.name == "x_amz_bedrock_agentcore_search":
            builtin_search_tool.append(tool)
    return builtin_search_tool
          
def execute_agent(
    user_prompt,
    model_id
):
    agent = create_react_agent(model_id, filter_search_tool())
    _response = asyncio.run(agent.ainvoke({
        "messages": user_prompt
    }))

    _response = _response.get('messages', {})[1].content
    print(
        f"Invoke Langchain Agents Response"
        f"Response - \n{_response}\n"
    )
    return _response