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}

다음 값을 교체합니다.

참고

에 대해 선택적으로 지원되는 파라미터 목록은 모델 컨텍스트 프로토콜 설명서의 리소스에서 요청 본문의 params 객체를 resources/list 참조하세요. 검색 창 옆의 페이지 상단에서 문서를 보려는 MCP 버전을 선택할 수 있습니다. 버전이 Amazon Bedrock AgentCore에서 지원하는 버전인지 확인합니다.

응답은 URIs, 이름, 설명 및 MIME 유형과 함께 사용 가능한 리소스 목록을 반환합니다.

참고

도구 및 프롬프트와 달리 리소스 URIs는 대상 이름 접두사 없이 반환됩니다. MCP 서버의 원래 URI는 있는 그대로 반환됩니다.

참고

여러 대상이 동일한 리소스 URI를 노출하면 게이트웨이는 가장 낮은 resourcePriority 값으로 대상에서 리소스를 반환합니다.

중요

리소스 URIs는 다운스트림 MCP 서버 대상에서 제공하며 게이트웨이에서 검증하거나 삭제하지 않습니다. 악의적이거나 손상된 MCP 서버는 내부 엔드포인트(SSRF) 또는 로컬 파일 시스템 경로(예: )를 가리키는 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}"))