Listing Tools with MCP Clients Calling Tools with MCP Clients Searching Tools with MCP Clients

Using MCP Client Libraries

Several client libraries are available to simplify working with MCP servers, including Gateway. These libraries provide high-level abstractions for listing tools, calling tools, and handling responses.

Listing Tools with MCP Clients

Here are examples of how to list tools using different MCP client libraries:

Note

If search is enabled on the gateway, then the search tool, x_amz_bedrock_agentcore_search will be listed first in the response.

Python with Requests


import requests
import json

def list_tools(gateway_url, access_token):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }
    
    payload = {
        "jsonrpc": "2.0",
        "id": "list-tools-request",
        "method": "tools/list"
    }
    
    response = requests.post(gateway_url, headers=headers, json=payload)
    return response.json()

# Example usage
gateway_url = "https://your-gateway-endpoint.execute-api.region.amazonaws.com/mcp"
access_token = "YOUR_ACCESS_TOKEN"
tools = list_tools(gateway_url, access_token)
print(json.dumps(tools, indent=2))

MCP Client



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. List available tools
            print("Listing tools...")
            cursor = True
            tools = []
            while cursor:
                next_cursor = cursor
                if type(cursor) == bool:
                    next_cursor = None
                list_tools_response = await session.list_tools(next_cursor)
                tools.extend(list_tools_response.tools)
                cursor = list_tools_response.nextCursor

            tool_names = []
            if tools:
                for tool in tools:
                    tool_names.append(tool.name)
            tool_names_string = "\n".join(tool_names)
            print(
                f"List MCP tools. # of tools - {len(tools)}"
               f"List of tools - \n{tool_names_string}\n"
            )

Strands MCP Client


from strands import Agent
import logging
from strands.models import BedrockModel
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client 
import os

def create_streamable_http_transport(mcp_url: str, access_token: str):
       return streamablehttp_client(mcp_url, headers={"Authorization": f"Bearer {access_token}"})


def get_full_tools_list(client):
    """
    List tools w/ support for pagination
    """
    more_tools = True
    tools = []
    pagination_token = None
    while more_tools:
        tmp_tools = client.list_tools_sync(pagination_token=pagination_token)
        tools.extend(tmp_tools)
        if tmp_tools.pagination_token is None:
            more_tools = False
        else:
            more_tools = True 
            pagination_token = tmp_tools.pagination_token
    return tools

def run_agent(mcp_url: str, access_token: str):
    mcp_client = MCPClient(lambda: create_streamable_http_transport(mcp_url, access_token))
     
    with mcp_client:
        tools = get_full_tools_list(mcp_client)
        print(f"Found the following tools: {[tool.tool_name for tool in tools]}")
        
run_agent(<MCP URL>, <Access token>)

LangGraph MCP Client



import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient


def list_tools(
    url,
    headers
):
    mcp_client = MultiServerMCPClient(
        {
            "agent": {
                "transport": "streamable_http",
                "url": url,
                "headers": headers,
            }
        }
    )
    tools = asyncio.run(mcp_client.get_tools())
    tool_details = []
    tool_names = []
    for tool in tools:
        tool_names.append(f"{tool.name}")
        tool_detail = f"{tool.name} - {tool.description} \n"

        tool_properties = tool.args_schema.get('properties', {})
        properties = []
        for property_name, tool_property in tool_properties.items():
            properties.append(f"{property_name} - {tool_property.get('description', None)} \n")
        tool_details.append(f"{tool_detail}{"\n".join(properties)}")

    tool_details_string = "\n".join(tool_details)
    tool_names_string = "\n".join(tool_names)
    print(
        f"Langchain: List MCP tools. # of tools - {len(tools)}\n",
        f"Langchain: List of tool names - \n{tool_names_string}\n"
        f"Langchain: Details of tools - \n{tool_details_string}\n"
    )

Calling Tools with MCP Clients

Here are examples of how to call tools using different MCP client libraries:

Python with Requests



import requests
import json

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

# Example usage
gateway_url = "https://your-gateway-endpoint.execute-api.region.amazonaws.com/mcp"
access_token = "YOUR_ACCESS_TOKEN"
result = call_tool(
    gateway_url, 
    access_token, 
    "getOrderStatus", 
    {"orderId": "ORD-12345-67890", "customerId": "CUST-98765"}
)
print(json.dumps(result, 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 a tool
            list_tools_response = await session.list_tools()
            tools = list_tools_response.tools
            print("Invoking Tools...")
            for tool in tools:
                tool_name = tool.name
                args = {
                    "param1": "paramValue1"
                }
                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

Note

This is for invoking agent



import functools

from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.tools.mcp.mcp_client import MCPClient


def execute_agent(
        bedrock_model,
        prompt
):
    mcp_client = MCPClient(functools.partial(_create_streamable_http_transport))
    tools = mcp_client.list_tools_sync()
    with mcp_client:
        agent = Agent(
            model=bedrock_model,
            tools=tools
        )
        return agent(prompt)


def _create_streamable_http_transport(
        url,
        headers=None
):
    return streamablehttp_client(
        url,
        headers=headers
    )

LangGraph MCP Client

Note

This is for invoking agent



import asyncio

from langgraph.prebuilt import create_react_agent


def execute_agent(
    user_prompt,
    model_id,
    tools
):
    agent = create_react_agent(model_id, tools)
    _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

Searching Tools with MCP Clients

Here are examples of how to search for tools using different MCP client libraries:

Python with Requests



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://your-gateway-endpoint.execute-api.region.amazonaws.com/mcp"
access_token = "YOUR_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 functools

from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.tools.mcp.mcp_client import MCPClient


def execute_agent(
        bedrock_model,
        prompt
):
    mcp_client = MCPClient(functools.partial(_create_streamable_http_transport))
    with mcp_client:
        agent = Agent(
            model=bedrock_model,
            tools=filter_search_tool(mcp_client)
        )
        return agent(prompt)


def filter_search_tool(
        mcp_client
):
    tools = mcp_client.list_tools_sync()
    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 _create_streamable_http_transport(
        url,
        headers=None
):
    return streamablehttp_client(
        url,
        headers=headers
    )

LangGraph MCP Client



import asyncio

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

url = ""
headers = {}
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


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

Warning Javascript is disabled or is unavailable in your browser.

To use the Amazon Web Services Documentation, Javascript must be enabled. Please refer to your browser's Help pages for instructions.

Document Conventions

Invoking a Tool

Testing your gateway