はじめに
Amazon Bedrock AgentCore考察記事第3弾。
前回までの記事では、Amazon Bedrock AgentCoreランタイムに主眼を当てて考察を行ってきた。
今回は、Amazon Bedrock AgentCore GatewayとIdentityに焦点を当てていく。
公式ブログによれば、Amazon Bedrock AgentCore Gatewayは以下の機能を提供してくれるフルサーバレスのマネージドサービスだ。書いてのとおり、既存のAPIをMCPに簡単に変換することも可能で、リモートMCPサーバとして実行できるため、共通的な機能をMCPサーバとして切り出して公開がしやすくなったと言える。
実際には、既存APIを「そのまま」MCPサーバとして公開するのは難しいと感じた。
HTTPレスポンスのステータスコードが、HTTPのコンテキストとMCPのコンテキストで異なるため、ゲートウェイ側での変換機能がないと、エージェントアプリケーションが思わぬ挙動をする可能性がある。
詳細は本編にて記載する。
AgentCore Gateway を使用すると、エージェントは、Smithy モデル、Lambda 関数、内部 API、および OpenAPI 仕様を使用するサードパーティープロバイダーを使用して AWS サービスにアクセスできます。同サービスは、着信リクエストとターゲットリソースに対する発信接続の両方のために安全なアクセスコントロールを実現することを目的として、二重認証モデルを採用しています。Lambda 関数は、外部システム (特に標準 API がないアプリケーションや、情報を取得するために複数のステップを実行する必要があるアプリケーション) を統合するために使用できます。
AgentCore Gateway は、認証、認可、スロットリング、カスタムリクエスト/レスポンス変換 (基盤となる API 形式に合わせるため)、マルチテナンシー、ツール選択など、同サービスがなければほとんどのお客様が独自に構築しなければならないであろう分野横断的な機能を簡単に使用できるようにします。
ツール選択機能は、特定のエージェントのタスクに最も適したツールを見つけるのに役立ちます。AgentCore Gateway は、これらすべてのツールにわたって統一された MCP インターフェイスを提供し、AgentCore Identity を使用して、AWS サービスのように OAuth を標準でサポートしていないツールに OAuth インターフェイスを提供します。
今回は、OpenAPI3.0で仕様が公開されているAPIをMCPに対応させることを想定して構成を作っていく。
いつも通り、せっかく作るからには再現度を上げるためにTerraformで自動化できるようにしておく。
なお、今回の記事ではAPIサーバとして、AWS Lambda Function URLsを使うことにしている。AWS Lambdaの場合、Amazon Bedrock AgentCore IdentityではIAM認証ができる(というか、HTTPエンドポイントを公開していないAWS Lambda関数の場合、IAM認証しかできないと言った方が正しい)が、今回はあくまでも、テキトーなHTTPエンドポイントをMCPサーバ化することを試してみたかったので、敢えて変な構成にしている。
実際に運用する際には、構成に合わせた最適な認証方法を選択しよう。
Auth0
Auth0については、基本は公式のDeveloper Guideに従って構築していく。
Resource Server
Auth0のコンソール上、アプリケーション -> API となっている部分のリソースが、auth0_resource_serverだ。
まずはここで、推奨設定値のアプリケーションをつくっていこう。
identifierは、一意な識別子であればドメインでなくても良い。今回はURN形式で作成をした。
auth0_resource_server_scopesでは、APIにどのような認可のためのスコープを持たせるかを設定している。
resource "auth0_resource_server" "amazon_bedrock_agentcore_gateway_auth" {
name = "Amazon Bedrock AgentCore Gateway Resource Server"
identifier = "urn:${local.auth0_resource_server_id}:api"
signing_alg = "RS256"
allow_offline_access = false
token_lifetime = 3600
skip_consent_for_verifiable_first_party_clients = true
}
resource "auth0_resource_server_scopes" "amazon_bedrock_agentcore_gateway_auth" {
resource_server_identifier = auth0_resource_server.amazon_bedrock_agentcore_gateway_auth.identifier
scopes {
name = "invoke:gateway"
}
scopes {
name = "read:gateway"
}
}
クライアント
文脈上分かりにくいが、ここで言うクライアントは、Auth0のクライアントアプリケーションのことで、MCPクライアントではない。auth0_clientでクライアントを作成し、auth0_client_grantで↑で作ったAPIとの紐づけと、このクライアントに許可するスコープを設定している。
auth0_client_credentialsは、クライアントシークレットの受け渡しに必要(このリソースでないと取得できない)ので作成しておく。
resource "auth0_client" "amazon_bedrock_agentcore_gateway_auth" {
name = "Amazon Bedrock AgentCore Gateway API"
description = "API for Amazon Bedrock AgentCore Gateway authorization"
app_type = "non_interactive"
custom_login_page_on = false
is_first_party = true
is_token_endpoint_ip_header_trusted = false
oidc_conformant = true
require_proof_of_possession = false
grant_types = [
"client_credentials",
]
jwt_configuration {
alg = "RS256"
lifetime_in_seconds = 3600
secret_encoded = false
}
refresh_token {
leeway = 0
token_lifetime = 31557600
rotation_type = "non-rotating"
expiration_type = "non-expiring"
}
}
resource "auth0_client_grant" "amazon_bedrock_agentcore_gateway_auth" {
client_id = auth0_client.amazon_bedrock_agentcore_gateway_auth.id
audience = auth0_resource_server.amazon_bedrock_agentcore_gateway_auth.identifier
scopes = [
"invoke:gateway",
"read:gateway",
]
}
resource "auth0_client_credentials" "amazon_bedrock_agentcore_gateway_auth" {
client_id = auth0_client.amazon_bedrock_agentcore_gateway_auth.id
authentication_method = "client_secret_post"
}
AWS Lambda
MCP Serverとして実行するAWS Lambdaも先につくっていこう。
周辺リソース
IAMやAmazon CloudWatch Logsは特に難しい要件は無いので普通に用意しておく。
※入力された内容に応じてJSONを返すだけなので、ログ記録用にAmazon CloudWatch Logsさえ書ければ良い。
resource "aws_cloudwatch_log_group" "lambda" {
name = "/aws/lambda/${local.lambda_function_name}"
retention_in_days = 3
}
resource "aws_iam_role" "lambda" {
name = local.iam_lambda_role_name
assume_role_policy = data.aws_iam_policy_document.lambda_assume.json
}
data "aws_iam_policy_document" "lambda_assume" {
statement {
effect = "Allow"
actions = [
"sts:AssumeRole",
]
principals {
type = "Service"
identifiers = [
"lambda.amazonaws.com",
]
}
}
}
resource "aws_iam_role_policy" "lambda_custom" {
name = local.iam_lambda_policy_name
role = aws_iam_role.lambda.id
policy = data.aws_iam_policy_document.lambda_custom.json
}
data "aws_iam_policy_document" "lambda_custom" {
statement {
effect = "Allow"
actions = [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents",
]
resources = [
aws_cloudwatch_log_group.lambda.arn,
"${aws_cloudwatch_log_group.lambda.arn}:*",
]
}
}
AWS Lambda関数
ここも難しいことはないので、AWS Lambda Function URLsとまとめて作っておこう。
data "archive_file" "bac_gateway_lambda" {
type = "zip"
output_path = "<一時ファイルの格納パス>/example.zip"
source {
filename = "example.py"
content = file("<AWS Lambdaのソース格納パス>/example.py")
}
}
resource "aws_lambda_function" "bac_gateway" {
depends_on = [
aws_cloudwatch_log_group.lambda,
]
function_name = local.lambda_function_name
filename = data.archive_file.bac_gateway_lambda.output_path
role = aws_iam_role.lambda.arn
handler = "example.lambda_handler"
source_code_hash = data.archive_file.bac_gateway_lambda.output_base64sha256
runtime = "python3.12"
memory_size = 128
timeout = 60
}
resource "aws_lambda_function_url" "bac_gateway" {
function_name = aws_lambda_function.bac_gateway.function_name
authorization_type = "NONE"
}
AWS Lamdbda関数の実行コード
今回の例として
- nameをinputとしてもらう
- age, departmentのプロパティを返す
- とりあえず動かしたいだけなので、対応するnameは
neruneruoとhogehogeoのみ
というシンプルなアプリを作る。
例として以下に示しているスクリプトはいテキトーな中身だが、一つ気を付けなければいけないのは 「リソースが見つからないからといって安易にステータスコード404を返してはいけない」 という点だ。
ステータスコード404を返すと、Strands Agentsのアプリケーションがストールする。
※MCPのプロトコルでは404はRESTの対象リソースが見つからない場合と意味が異なるため、らしい。
このため、エラーになったかどうかの判別をできるように、レスポンスのBodyの中にisError: True/Falseを入れておく。
404応答のときはステータスコードを返したくなるのをグッと我慢して200応答を返そう。
つまり、Amazon Bedrock AgentCore Gatewayで既存のREST APIを変換する場合、多くのケースでそのままの利用はできないということになる。通常のREST APIは、ステータスコード404は対象のリソースが見つからない時に返すからだ。
さすがにこれは不便なので、GAまでに仕様が見直されることを期待したい。
import json
def lambda_handler(event, context):
request_path = event.get("requestContext", {}).get("http", {}).get("path")
request_method = event.get("requestContext", {}).get("http", {}).get("method")
request_query_string = event.get("queryStringParameters", {})
if request_path == "/profile":
if request_method == "GET":
if request_query_string.get("name") == "neruneruo":
return {
"statusCode": 200,
"body": json.dumps(
{
"isError": False,
"age": 30,
"department": "development",
}
),
}
elif request_query_string.get("name") == "hogehogeo":
return {
"statusCode": 200,
"body": json.dumps(
{
"isError": False,
"age": 45,
"department": "sales",
}
),
}
else:
print(f"return 404. input name: {request_query_string.get("name")}")
return {
"statusCode": 200,
"body": json.dumps(
{
"isError": True,
"message": "Person not found.",
}
),
}
else:
print(f"return 400. input method: {request_method}")
return {
"statusCode": 400,
"body": json.dumps(
{
"isError": True,
"message": "Invalid request method.",
}
),
}
else:
print(f"return 400. input path: {request_path}")
return {
"statusCode": 400,
"body": json.dumps(
{
"isError": True,
"message": "Invalid request path.",
}
),
}
Amazon S3
今回、AWS Lambda Function URLsをMCPサーバとしてアダプタするために、APIのインタフェースの仕様(OpenAPI準拠のYAML)を作ってAmazon S3に格納しておこう。ここも特に難易度は高くない。
example.jsonは、先ほど書いたようにステータスコード404応答の代わりにスキーマのプロパティにもisError属性を入れておく。
resource "aws_s3_bucket" "example" {
bucket = local.s3_bucket_name
force_destroy = true
}
resource "aws_s3_bucket_ownership_controls" "example" {
bucket = aws_s3_bucket.example.id
rule {
object_ownership = "BucketOwnerEnforced"
}
}
resource "aws_s3_bucket_public_access_block" "example" {
bucket = aws_s3_bucket.example.id
block_public_acls = false
block_public_policy = false
ignore_public_acls = false
restrict_public_buckets = false
}
locals {
rendered = templatefile("<スキーマファイルへのローカルパス>/example.json", {
lambda_bac_gateway_function_url = aws_lambda_function_url.bac_gateway.function_url,
})
}
resource "aws_s3_object" "example" {
bucket = aws_s3_bucket.example.id
key = "example.json"
content_type = "application/json"
content = local.rendered
etag = md5(local.rendered)
}
{
"openapi": "3.0.0",
"info": {
"title": "Get profile of person API",
"version": "1.0.0",
"description": "Get profile of person"
},
"servers": [
{
"url": "${lambda_bac_gateway_function_url}"
}
],
"paths": {
"/profile": {
"get": {
"summary": "Get profile of person",
"operationId": "getProfile",
"parameters": [
{
"name": "name",
"in": "query",
"description": "Name of person for searching profile",
"required": true,
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "Successful response",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"isError": {
"type": "boolean",
"description": "Whether an error occurred."
},
"age": {
"type": "number",
"description": "Age of person for searching profile. If isError equals true, this property is omitted."
},
"department": {
"type": "string",
"description": "Affiliated department of person for searching profile. If isError equals true, this property is omitted."
},
"message": {
"type": "string",
"description": "Details if isError equals true. If isError equals false, this property is omitted."
}
}
}
}
}
},
"400": {
"description": "Bad request",
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "Detailed error message"
}
}
}
}
}
}
}
}
}
}
}
さて、ここまでくれば事前準備は完了だ。
Amazon Bedrock AgentCore Gateway/Identity
Amazon Bedrock AgentCore GatewayでMCPサーバを作成する場合、構成要素は以下の3つだ
- Amazon Bedrock AgentCore OAuth2 Credential Provider: 認証情報
- Amazon Bedrock AgentCore Gateway: HTTPサーバ
- Amazon Bedrock AgentCore Gateway Target: 受信したMCPリクエストの転送先
それぞれのリソースについて解説していく。
IAM
まずは、Amazon Bedrock AgentCore Gatewayのサービス用IAMロールを作成する。
bedrock-agentcore:*Gateway*な情報が書いてあるステートメントは、以下の公式のDeveloper Guideを参考にしながら設定した。
また、bedrock-agentcore:GetResourceOauth2Tokenおよびbedrock-agentcore:GetWorkloadAccessTokenについては、自力で一つ一つ権限を確かめたので、おそらくこれが最小権限構成であると考えられる。
あとは、スキーマ取得のためのAmazon S3へのアクセス権限と、この後作成する、Auth0のclient_secretを保存するAWS Secrets Managerへのアクセス権限を付与しておけば良い。
resource "aws_iam_role" "bac_gateway" {
name = local.iam_bac_gateway_role_name
assume_role_policy = data.aws_iam_policy_document.bac_gateway_assume.json
}
data "aws_iam_policy_document" "bac_gateway_assume" {
statement {
effect = "Allow"
principals {
type = "Service"
identifiers = [
"bedrock-agentcore.amazonaws.com",
]
}
actions = [
"sts:AssumeRole",
]
}
}
resource "aws_iam_role_policy" "bac_gateway_custom" {
name = local.iam_bac_gateway_policy_name
role = aws_iam_role.bac_gateway.id
policy = data.aws_iam_policy_document.bac_gateway_custom.json
}
data "aws_iam_policy_document" "bac_gateway_custom" {
statement {
effect = "Allow"
actions = [
"bedrock-agentcore:ListGateways",
"bedrock-agentcore:GetGateway",
"bedrock-agentcore:ListGatewayTargets",
"bedrock-agentcore:GetGatewayTarget",
]
resources = [
"arn:aws:bedrock-agentcore:*:*:*gateway*",
]
}
statement {
effect = "Allow"
actions = [
"bedrock-agentcore:GetResourceOauth2Token",
"bedrock-agentcore:GetWorkloadAccessToken",
]
resources = [
"arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.self.id}:token-vault/default",
data.external.bedrock_agentcore_oauth2_credential_provider.result.credential_provider_arn,
"arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.self.id}:workload-identity-directory/default",
"arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.self.id}:workload-identity-directory/default/workload-identity/${local.bac_gateway_name}*",
]
}
statement {
effect = "Allow"
actions = [
"s3:GetObject",
]
resources = [
"${aws_s3_bucket.example.arn}/*",
]
}
statement {
effect = "Allow"
actions = [
"secretsmanager:GetSecretValue",
]
resources = [
data.external.bedrock_agentcore_oauth2_credential_provider.result.client_secret_arn,
]
}
}
Amazon Bedrock AgentCore OAuth2 Credential Provider
Amazon Bedrock AgentCore OAuth2 Credential Providerは、Auth0に対して認証リクエストを送ってくれるリソースだ。
discovery_urlはAuth0が固定で提供するドメインなので、記載の通りで特に考えることはない。
※discovery_urlを使うことでトークン取得APIのエンドポイントの情報を入力せずとも、このURLから情報が取得できる
あとは、Auth0作成時に作られたクライアントIDとシークレットを設定する。
クライアントシークレットを設定すると、AWS Secrets Managerのリソースが勝手に作られる。
先回りして作ることはできないが、このリソースの削除と同時に消えてくれるため、あまり気にしなくて良い。
data "external" "bedrock_agentcore_oauth2_credential_provider" {
program = ["python3", "./agentcore_oauth2_credential_provider.py"]
query = {
aws_region = data.aws_region.current.region
provider_name = local.bac_oauth2_credential_provider_name
discovery_url = "https://${var.auth0_domain}/.well-known/openid-configuration"
client_id = auth0_client.amazon_bedrock_agentcore_gateway_auth.client_id
client_secret = auth0_client_credentials.amazon_bedrock_agentcore_gateway_auth.client_secret
}
}
"""
Create Amazon Bedrock AgentCore OAuth2 Credential Provider if it doesn't exist yet.
"""
import boto3
import botocore
import json
import sys
def main():
params = json.load(sys.stdin)
aws_region = params["aws_region"]
provider_name = params["provider_name"]
discovery_url = params["discovery_url"]
client_id = params["client_id"]
client_secret = params["client_secret"]
bac = boto3.client("bedrock-agentcore-control", region_name=aws_region)
try:
get_result = bac.get_oauth2_credential_provider(name=provider_name)
print(
json.dumps(
{
"credential_provider_arn": get_result["credentialProviderArn"],
"client_secret_arn": get_result["clientSecretArn"]["secretArn"],
}
)
)
exit(0)
except botocore.exceptions.ClientError as e:
code = e.response.get("Error", {}).get("Code")
if code == "ResourceNotFoundException":
pass
else:
sys.stderr.write(
json.dumps(
{
"error": f"get_oauth2_credential_provider() failed {str(e)} {code}"
}
)
)
sys.exit(1)
try:
create_result = bac.create_oauth2_credential_provider(
name=provider_name,
credentialProviderVendor="CustomOauth2",
oauth2ProviderConfigInput={
"customOauth2ProviderConfig": {
"oauthDiscovery": {
"discoveryUrl": discovery_url,
},
"clientId": client_id,
"clientSecret": client_secret,
},
},
)
except botocore.exceptions.ClientError as e:
sys.stderr.write(
json.dumps(
{"error": f"create_oauth2_credential_provider() failed {str(e)}"}
)
)
sys.exit(1)
print(
json.dumps(
{
"credential_provider_arn": create_result["credentialProviderArn"],
"client_secret_arn": create_result["clientSecretArn"]["secretArn"],
}
)
)
if __name__ == "__main__":
main()
Amazon Bedrock AgentCore Gateway
ゲートウェイは以下のように作成する。
例によって、スクリプト内はリソースが存在した場合はその情報を返し、無い場合はcreateして状態が安定してから情報を返すようにしている。
discovery_urlはAmazon Bedrock AgentCore OAuth2 Credential Providerと同じ設定を入れれば良い。
audienceは、Auth0のAPIを作成する時に設定したidentifierの値を設定する。
gateway_nameはAPIの長さ制限にかからなくても、ドメイン名が長くなりすぎて名前解決が出来なくなることがあるので、適切な長さを設定しよう。
data "external" "bedrock_agentcore_gateway" {
program = ["python3", "./agentcore_gateway.py"]
query = {
aws_region = data.aws_region.current.region
gateway_name = local.bac_gateway_name
role_arn = aws_iam_role.bac_gateway.arn
discovery_url = "https://${var.auth0_domain}/.well-known/openid-configuration"
audience = auth0_resource_server.amazon_bedrock_agentcore_gateway_auth.identifier
}
}
"""
Create Amazon Bedrock AgentCore Gateway if it doesn't exist yet.
"""
import boto3
import botocore
import json
import sys
import time
def main():
params = json.load(sys.stdin)
aws_region = params["aws_region"]
gateway_name = params["gateway_name"]
role_arn = params["role_arn"]
discovery_url = params["discovery_url"]
audience = params["audience"]
bac = boto3.client("bedrock-agentcore-control", region_name=aws_region)
try:
paginator = bac.get_paginator("list_gateways")
for page in paginator.paginate():
for gateway in page.get("items", []):
if gateway.get("name") == gateway_name:
get_result = bac.get_gateway(gatewayIdentifier=gateway["gatewayId"])
print(
json.dumps(
{
"gateway_id": get_result["gatewayId"],
"gateway_arn": get_result["gatewayArn"],
"gateway_url": get_result["gatewayUrl"],
}
)
)
sys.exit(0)
except botocore.exceptions.ClientError as e:
sys.stderr.write(str(e))
print(json.dumps({"error": "list_gateways()/get_gateway() failed."}))
sys.exit(1)
try:
create_result = bac.create_gateway(
name=gateway_name,
roleArn=role_arn,
protocolType="MCP",
authorizerType="CUSTOM_JWT",
authorizerConfiguration={
"customJWTAuthorizer": {
"discoveryUrl": discovery_url,
"allowedAudience": [
audience,
],
},
},
exceptionLevel="DEBUG",
)
except botocore.exceptions.ClientError as e:
sys.stderr.write(json.dumps({"error": f"create_gateway() failed {str(e)}"}))
sys.exit(1)
try:
while True:
get_result = bac.get_gateway(gatewayIdentifier=create_result["gatewayId"])
if get_result["status"] == "READY":
break
elif get_result["status"] in ("FAILED"):
raise RuntimeError(f"invalid status {get_result["status"]}.")
time.sleep(1)
except botocore.exceptions.ClientError as e:
sys.stderr.write(json.dumps({"error": f"get_gateway() failed {str(e)}"}))
sys.exit(1)
except RuntimeError as e:
sys.stderr.write(json.dumps({"error": str(e)}))
sys.exit(1)
print(
json.dumps(
{
"gateway_id": create_result["gatewayId"],
"gateway_arn": create_result["gatewayArn"],
"gateway_url": create_result["gatewayUrl"],
}
)
)
if __name__ == "__main__":
main()
Amazon Bedrock AgentCore Gateway Target
このリソースが転送先を設定するものだ。
AWS Lambda Function URLsのURLどこ？と思われるかもしれないが、Amazon S3に格納しているスキーマの中に埋め込んでいるため、そこにアクセスしに行くようになっている。
credential_provider_arnは先に作ったAmazon Bedrock AgentCore OAuth2 Credential ProviderのARNを指せばよい。
MCPサーバのツール名は[gateway_target_name]___[スキーマで指定したoperationId]が設定される。
gateway_target_nameを長くし過ぎると、Tool_Useのツール名が長くなりすぎてエラーになるため、Amazon Bedrock AgentCore Gateway同様、適切な長さを設定しよう。
問題はaudienceで、Developer Guideには記載がないが、credentialProviderConfigurations[0].credentialProvider.oauthCredentialProvider.customParametersにこの設定をしないとAuth0に対するトークン検証のリクエストがエラーになる。しかも、 なぜかマネージメントコンソール上でこの設定値が見えない(テキストボックスがあるのに設定値が表示されない) ので、設定が漏れないよう注意が必要だ。
data "external" "bedrock_agentcore_gateway_target" {
program = ["python3", "./agentcore_gateway_target.py"]
query = {
aws_region = data.aws_region.current.region
gateway_id = data.external.bedrock_agentcore_gateway.result.gateway_id
gateway_target_name = local.bac_gateway_target_name
s3_uri = "s3://${aws_s3_object.example.id}"
credential_provider_arn = data.external.bedrock_agentcore_oauth2_credential_provider.result.credential_provider_arn
audience = auth0_resource_server.amazon_bedrock_agentcore_gateway_auth.identifier
}
}
"""
Create Amazon Bedrock AgentCore Gateway Target if it doesn't exist yet.
"""
import boto3
import botocore
import json
import sys
import time
def main():
params = json.load(sys.stdin)
aws_region = params["aws_region"]
gateway_id = params["gateway_id"]
gateway_target_name = params["gateway_target_name"]
s3_uri = params["s3_uri"]
credential_provider_arn = params["credential_provider_arn"]
audience = params["audience"]
bac = boto3.client("bedrock-agentcore-control", region_name=aws_region)
try:
paginator = bac.get_paginator("list_gateway_targets")
for page in paginator.paginate(gatewayIdentifier=gateway_id):
for item in page.get("items", []):
if item.get("name") == gateway_target_name:
get_result = bac.get_gateway_target(
gatewayIdentifier=gateway_id,
targetId=item["targetId"],
)
print(
json.dumps(
{
"target_id": item["targetId"],
}
)
)
sys.exit(0)
except botocore.exceptions.ClientError as e:
sys.stderr.write(str(e))
print(json.dumps({"error": "list_agent_runtimes() failed."}))
sys.exit(1)
try:
create_result = bac.create_gateway_target(
gatewayIdentifier=gateway_id,
name=gateway_target_name,
targetConfiguration={
"mcp": {
"openApiSchema": {"s3": {"uri": s3_uri}},
},
},
credentialProviderConfigurations=[
{
"credentialProviderType": "OAUTH",
"credentialProvider": {
"oauthCredentialProvider": {
"providerArn": credential_provider_arn,
"scopes": [
"invoke:gateway",
"read:gateway",
],
"customParameters": {"audience": audience},
},
},
},
],
)
except botocore.exceptions.ClientError as e:
sys.stderr.write(
json.dumps({"error": f"create_gateway_target() failed {str(e)}"})
)
sys.exit(1)
try:
while True:
get_result = bac.get_gateway_target(
gatewayIdentifier=gateway_id,
targetId=create_result["targetId"],
)
if get_result["status"] == "READY":
break
elif get_result["status"] in ("FAILED"):
raise RuntimeError(f"invalid status {get_result["status"]}.")
time.sleep(1)
except botocore.exceptions.ClientError as e:
sys.stderr.write(json.dumps({"error": f"get_gateway_target() failed {str(e)}"}))
sys.exit(1)
except RuntimeError as e:
sys.stderr.write(json.dumps({"error": str(e)}))
sys.exit(1)
print(
json.dumps(
{
"target_id": create_result["targetId"],
}
)
)
if __name__ == "__main__":
main()
ここまで作ったら、terraform applyでリソースを作ろう。
いざ、動かす！
動かすためのクライアントをStrands Agentsで作っておこう。
普段のエージェントに、こんな感じで付け加えれば良い。
from strands.tools.mcp.mcp_client import MCPClient
# (中略)
def create_streamable_http_transport():
return streamablehttp_client(
url=os.environ.get("AWS_BEDROCK_AGENTCORE_GATEWAY_URL"),
headers={"Authorization": f"Bearer {os.environ.get("BEARER_ACCESS_TOKEN")}"},
)
mcp_client = MCPClient(create_streamable_http_transport)
# (中略)
with mcp_client:
try:
strands_agents_tools = mcp_client.list_tools_sync()
agent = Agent(
model=bedrock_model,
system_prompt="You are professional of getting personal information.You can absolutely trust input names",
tools=strands_agents_tools,
)
# (以下略)
さらに、BEARER_ACCESS_TOKENの環境変数に設定するために、Auth0にトークンリクエストをしよう。
一番手っ取り早いのはcurlだ。Auth0のコンソールにログインして、左メニューの「アプリケーション→アプリケーション」でこの画面を出し、クイックスタートのタブの中をコピペする。
これで、Strands Agentを動かすと、以下のように表示される。
$ python3 main.py "neruneruoについて教えて"
ご質問いただきありがとうございます。「neruneruoさん」についての情報を調べるためには、プロフィール検索機能を使用できます。名前を使ってプロフィール情報を取得しましょう。
Tool #1: xxx-mcp-example-gateway-target___getProfile
neruneruoさんのプロフィール情報は以下の通りです：
- 年齢: 30歳
- 部署: 開発部門 (development)
これが現在把握できるneruneruoさんの基本情報です。他に何か特定の情報をお知りになりたい場合は、お気軽にお尋ねください。
動いた！ちゃんとTool #1で作成したMCPサーバにアクセスしている！
さらに、Not Foundになるような名前で検索をかけると、
$ python3 main.py "neruについて教えて"
NEEDsについて情報を取得するには、プロフィールを検索する必要があります。「neru」という名前で検索してみましょう。
Tool #1: xxx-mcp-example-gateway-target___getProfile
申し訳ありませんが、「neru」という名前の人物のプロフィール情報は見つかりませんでした。
正確な名前のスペルや表記が異なる可能性があります。別の名前や正確なスペルで検索したい場合は、お知らせください。
となり、ちゃんとisError: trueが効いていそうだ。
これで、お手軽にMCPサーバを作れるようになったぞ！