[アップデート] Amazon Bedrock AgentCore GatewayのターゲットにAmazon API Gatewayが追加されました

"""Cognito Client Credentials Flow でアクセストークンを取得するスクリプト"""

import os
import base64
import requests

# 設定
TOKEN_ENDPOINT = "https://xxx.auth.us-west-2.amazoncognito.com/oauth2/token"
CLIENT_ID = os.environ.get("COGNITO_CLIENT_ID", "xxx")
CLIENT_SECRET = os.environ.get("COGNITO_CLIENT_SECRET")

def get_access_token():
    if not CLIENT_SECRET:
        raise ValueError("環境変数 COGNITO_CLIENT_SECRET を設定してください")

    credentials = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()

    headers = {
        "Authorization": f"Basic {credentials}",
        "Content-Type": "application/x-www-form-urlencoded"
    }

    data = {
        "grant_type": "client_credentials"
    }

    response = requests.post(TOKEN_ENDPOINT, headers=headers, data=data)
    response.raise_for_status()

    return response.json()

export CLIENT_ID=xxx
export CLIENT_SECRET=yyy

"""Bedrock AgentCore Gateway からツール一覧を取得するスクリプト"""

import json
import os

import requests

from get_cognito_token import get_access_token

GATEWAY_URL = "https://gateway-quick-start-xxx.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"

def list_tools(gateway_url: str, bearer_token: str):
    """List all available tools from the gateway"""
    response = requests.post(
        gateway_url,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {bearer_token}",
        },
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list",
            "params": {},
        },
    )

    print(f"Status Code: {response.status_code}")
    print(f"Response Body: {json.dumps(response.json(), indent=2)}")
    return response

if __name__ == "__main__":
    if not GATEWAY_URL:
        raise ValueError("環境変数 GATEWAY_URL を設定してください")

    # Cognitoからアクセストークン取得
    token_response = get_access_token()
    access_token = token_response.get("access_token")

    # ツール一覧取得
    list_tools(GATEWAY_URL, access_token)

python3 list_tools.py

# 実行結果
Status Code: 200
Response Body: {
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "inputSchema": {
          "type": "object",
          "properties": {
            "basePath": {
              "type": "string"
            },
            "price": {
              "type": "number"
            },
            "type": {
              "type": "string",
              "enum": [
                "dog",
                "cat",
                "fish",
                "bird",
                "gecko"
              ]
            }
          }
        },
        "name": "target-quick-start-xxx___CreatePet",
        "description": "CreatePet"
      },
      {
        "inputSchema": {
          "type": "object",
          "properties": {
            "basePath": {
              "type": "string"
            },
            "page": {
              "type": "string"
            },
            "type": {
              "type": "string"
            }
          }
        },
        "name": "target-quick-start-xxx___GetAllPets",
        "description": "GetAllPets"
      },
      {
        "inputSchema": {
          "type": "object",
          "properties": {
            "petId": {
              "type": "string"
            },
            "basePath": {
              "type": "string"
            }
          },
          "required": [
            "petId"
          ]
        },
        "name": "target-quick-start-xxx___GetPet",
        "description": "GetPet"
      }
    ]
  }
}

#!/usr/bin/env python3
"""Bedrock AgentCore Gateway の GetAllPets ツールを呼び出すスクリプト"""

import json
import os

import requests

from get_cognito_token import get_access_token

GATEWAY_URL = "https://gateway-quick-start-xxx.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"

def call_tool(
    gateway_url: str, bearer_token: str, tool_name: str, arguments: dict = None
):
    """Call a specific tool on the gateway"""
    response = requests.post(
        gateway_url,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {bearer_token}",
        },
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": arguments or {},
            },
        },
    )

    print(f"Status Code: {response.status_code}")
    print(f"Response Body: {json.dumps(response.json(), indent=2)}")
    return response

def get_all_pets(
    gateway_url: str, bearer_token: str, page: str = None, pet_type: str = None
):
    """GetAllPets ツールを呼び出す"""
    arguments = {}
    if page:
        arguments["page"] = page
    if pet_type:
        arguments["type"] = pet_type

    return call_tool(
        gateway_url, bearer_token, "target-quick-start-xxx___GetAllPets", arguments
    )

if __name__ == "__main__":
    if not GATEWAY_URL:
        raise ValueError("環境変数 GATEWAY_URL を設定してください")

    # Cognitoからアクセストークン取得
    token_response = get_access_token()
    access_token = token_response.get("access_token")

    # GetAllPets 呼び出し
    get_all_pets(GATEWAY_URL, access_token)

python3 call_get_all_pets.py

# 実行結果
Status Code: 200
Response Body: {
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "isError": false,
    "content": [
      {
        "type": "text",
        "text": "[{\"id\":1,\"type\":\"dog\",\"price\":249.99},{\"id\":2,\"type\":\"cat\",\"price\":124.99},{\"id\":3,\"type\":\"fish\",\"price\":0.99}]"
      }
    ]
  }
}