View a markdown version of this page

列出 AgentCore 网关中的可用资源 - 亚马逊基岩 AgentCore

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

列出 AgentCore 网关中的可用资源

要列出 AgentCore 网关提供的所有可用资源,请向网关的 MCP 端点发出 POST 请求,并在请求正文中指定resources/list为方法:

例
2025-11-25 and earlier
POST /mcp HTTP/1.1 Host: ${GatewayEndpoint} Accept: application/json, text/event-stream Content-Type: application/json Authorization: ${Authorization header} MCP-Protocol-Version: ${McpProtocolVersion} ${RequestBody}
2026-07-28

在版本上2026-07-28,每个请求都带有MCP-Protocol-Version标头、Mcp-Method请求元数据标头和正_meta文中的版本字段。

POST /mcp HTTP/1.1 Host: ${GatewayEndpoint} Accept: application/json, text/event-stream Content-Type: application/json Authorization: ${Authorization header} MCP-Protocol-Version: 2026-07-28 Mcp-Method: resources/list ${RequestBody}
注意

网关仅接受其protocolConfiguration.mcp配置supportedVersions字段中列出的 MCP 协议版本。要使用版本2026-07-28,请确保您的网关supportedVersions包含该版本。您可以使用 UpdateGateway API 更改支持的版本。

替换以下值:

注意

有关可选支持的参数列表resources/list,请参阅模型上下文协议文档资源中请求正文中的params对象。在页面顶部的搜索栏旁边,您可以选择要查看其文档的 MCP 版本。确保该版本是亚马逊 Bedrock AgentCore 支持的版本。

响应返回可用资源的列表,包括其 URI、名称、描述和 MIME 类型。

注意

与工具和提示不同,返回的资源 URI 不带目标名称前缀。来自 MCP 服务器的原始 URI 按原样返回。

注意

当多个目标暴露相同的资源 URI 时,网关会从目标返回resourcePriority值最低的资源。

重要

资源 URI 由下游 MCP 服务器目标提供,未经网关验证或清理。恶意或受感染的 MCP 服务器可能会返回指向内部端点 (SSRF) 或本地文件系统路径(例如)的 URI。file:///etc/passwd在关注资源 URI 之前对其进行验证和清理,不要自动从不可信的 MCP 服务器目标获取或呈现 URI。

列出资源的代码示例

要查看在网关中列出可用资源的示例,请选择以下方法之一:

例
Python requests package (2025-11-25 and earlier)

将标MCP-Protocol-Version头设置为您的网关支持的版本。

import requests import json def list_resources(gateway_url, access_token): headers = { "Content-Type": "application/json", "Authorization": f"Bearer {access_token}", "MCP-Protocol-Version": "2025-11-25" } payload = { "jsonrpc": "2.0", "id": "list-resources-request", "method": "resources/list" } response = requests.post(gateway_url, headers=headers, json=payload) return response.json() # Example usage gateway_url = "https://${GatewayEndpoint}/mcp" # Replace with your actual gateway endpoint access_token = "${AccessToken}" # Replace with your actual access token resources = list_resources(gateway_url, access_token) print(json.dumps(resources, indent=2))
Python requests package (2026-07-28)

在版本上2026-07-28,添加MCP-Protocol-Version和Mcp-Method标头,并在请求正文中包含_meta版本字段。您的网关supportedVersions必须包括2026-07-28。

import requests import json def list_resources(gateway_url, access_token): headers = { "Content-Type": "application/json", "Authorization": f"Bearer {access_token}", "MCP-Protocol-Version": "2026-07-28", "Mcp-Method": "resources/list" } payload = { "jsonrpc": "2.0", "id": "list-resources-request", "method": "resources/list", "params": { "_meta": { "io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientInfo": {"name": "my-agent", "version": "1.0.0"}, "io.modelcontextprotocol/clientCapabilities": {} } } } response = requests.post(gateway_url, headers=headers, json=payload) return response.json() # Example usage gateway_url = "https://${GatewayEndpoint}/mcp" # Replace with your actual gateway endpoint access_token = "${AccessToken}" # Replace with your actual access token resources = list_resources(gateway_url, access_token) print(json.dumps(resources, indent=2))
MCP Client
import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def execute_mcp( url, token, headers=None ): default_headers = { "Authorization": f"Bearer {token}" } headers = {**default_headers, **(headers or {})} 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 resources print("Listing resources...") cursor = True resources = [] while cursor: next_cursor = cursor if isinstance(cursor, bool): next_cursor = None list_resources_response = await session.list_resources(next_cursor) resources.extend(list_resources_response.resources) cursor = list_resources_response.nextCursor if resources: for resource in resources: print(f"{resource.uri} - {resource.name} ({resource.mimeType})") print(f"Total resources: {len(resources)}") async def main(): url = "https://${GatewayEndpoint}/mcp" token = "your_bearer_token_here" # Optional additional headers additional_headers = { "Content-Type": "application/json", } await execute_mcp( url=url, token=token, headers=additional_headers ) # Run the async function if __name__ == "__main__": asyncio.run(main())
Strands MCP Client
# NOTE: Strands SDK resource support may vary. Use the MCP Client approach # above for the most reliable resources/list implementation. from strands.tools.mcp.mcp_client import MCPClient from mcp.client.streamable_http import streamablehttp_client def create_streamable_http_transport(mcp_url: str, access_token: str): return streamablehttp_client(mcp_url, headers={"Authorization": f"Bearer {access_token}"}) def run_agent(mcp_url: str, access_token: str): mcp_client = MCPClient(lambda: create_streamable_http_transport(mcp_url, access_token)) with mcp_client: result = mcp_client.list_resources_sync() print(f"Found the following resources: {[r.uri for r in result.resources]}") run_agent(<MCP URL>, <Access token>)
LangGraph MCP Client
# NOTE: LangGraph MCP adapter resource support may vary. Use the MCP Client # approach above for the most reliable resources/list implementation. import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def list_resources(url, token): headers = {"Authorization": f"Bearer {token}"} async with streamablehttp_client(url=url, headers=headers) as ( read_stream, write_stream, callA ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() response = await session.list_resources() for resource in response.resources: print(f"{resource.uri} - {resource.name}") asyncio.run(list_resources("https://${GatewayEndpoint}/mcp", "${AccessToken}"))