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}

取代以下的值：

  • ${GatewayEndpoint} – 閘道的 URL，如 CreateGateway API 的回應所提供。

  • ${Authorization header} – 當您設定傳入授權時，來自身分提供者的授權憑證。

  • ${RequestBody} – 請求內文的 JSON 承載，如模型內容通訊協定 (MCP) 中的列出資源所指定。包含 resources/list做為 method。

注意

如需 resources/list 選用支援參數的清單，請參閱模型內容通訊協定文件 中 資源請求內文中的 params 物件。在搜尋列旁的頁面頂端，您可以選取要檢視其文件的 MCP 版本。請確定 Amazon Bedrock AgentCore 支援該版本。

回應會傳回可用資源清單及其 URIs、名稱、描述和 MIME 類型。

注意

與工具和提示不同，傳回的資源 URIs沒有目標名稱字首。來自 MCP 伺服器的原始 URI 會依原狀傳回。

注意

當多個目標公開相同的資源 URI 時，閘道會從具有最低resourcePriority值的目標傳回資源。

重要

資源 URIs由下游 MCP 伺服器目標提供，且未經閘道驗證或淨化。惡意或遭入侵的 MCP 伺服器可能會傳回URIs （例如 file:///etc/passwd)。在遵循資源 URIs 之前進行驗證和清理，並且不會自動從不受信任的 MCP 伺服器目標擷取或轉譯 URIs。

列出資源的程式碼範例

若要查看閘道中列出可用資源的範例，請選取下列其中一種方法：

範例
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}"))