View a markdown version of this page

列出 AgentCore 网关中的可用资源 - Amazon Bedrock AgentCore
列出资源的代码示例

列出 AgentCore 网关中的可用资源

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

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

${RequestBody}

替换以下值：

注意

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

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

注意

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

注意

当多个目标公开相同的资源 URI 时，网关会从具有最低resourcePriority值的目标返回资源。

重要

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

列出资源的代码示例

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

例
Python requests package

  1. import requests
import json

def list_resources(gateway_url, access_token):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {access_token}"
    }

    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))
MCP Client

  1. 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

  1. # 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

  1. # 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}"))