Amazon Bedrock AgentCore is in preview release and is subject to change.
Call a tool in a AgentCore gateway
To call a specific tool, make a POST request to the gateway's MCP endpoint and specify tools/call
as the method in the request body, name of the tool, and the arguments:
POST /mcp HTTP/1.1
Host: ${GatewayEndpoint}
Content-Type: application/json
Authorization: Bearer ${AccessToken}
${RequestBody}
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.
-
${RequestBody}
– The JSON payload of the request body, as specified in Calling tools in the Model Context Protocol (MCP). Include tools/call
as the method
and include the name
of the tool and its arguments
.
The response returns the content returned by the tool and associated metadata.
Code samples for calling tools
To see examples of listing available tools in the gateway, select one of the following methods:
- curl
-
The following curl request shows an example request to call a tool called searchProducts
through a gateway with the ID mygateway-abcdefghij
.
curl -X POST \
https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \
-H "Content-Type: application/json" \
-H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
-d '{
"jsonrpc": "2.0",
"id": "invoke-tool-request",
"method": "tools/call",
"params": {
"name": "searchProducts",
"arguments": {
"query": "wireless headphones",
"category": "Electronics",
"maxResults": 2,
"priceRange": {
"min": 50.00,
"max": 200.00
}
}
}
}'
- Python requests package
-
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://${GatewayEndpoint}
" # Replace with your actual gateway endpoint
access_token = "${AccessToken}"
# Replace with your actual 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
-
This is for invoking agent
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
-
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
Errors
The tools/call
operation can return the following types of errors:
-
Errors returned as part of the HTTP status code:
- AuthenticationError
-
The request failed due to invalid authentication credentials.
HTTP Status Code: 401
- AuthorizationError
-
The caller does not have permission to invoke the tool.
HTTP Status Code: 403
- ResourceNotFoundError
-
The specified tool does not exist.
HTTP Status Code: 404
- ValidationError
-
The provided arguments do not conform to the tool's input schema.
HTTP Status Code: 400
- ToolExecutionError
-
An error occurred while executing the tool.
HTTP Status Code: 500
- InternalServerError
-
An internal server error occurred.
HTTP Status Code: 500
-
MCP errors. For more information about these types of errors, Error Handling in the Model Context Protocol (MCP) documentation.