[Amazon Bedrock AgentCore] Gateway経由でLambda関数をAIエージェントのツールにしてみた

gateway_response = gateway_client.create_mcp_gateway( name=gateway_name, role_arn=gateway_role_arn, authorizer_config=authorizer_config, enable_semantic_search=True # セマンティック検索を有効化！ ) 

async def search_tools(gateway_url, access_token, query): headers = {"Authorization": f"Bearer {access_token}"} async with streamablehttp_client(gateway_url, headers=headers) as (read, write, _): async with ClientSession(read, write) as session: await session.initialize() # Use the built-in search tool response = await session.call_tool( "x_amz_bedrock_agentcore_search", arguments={"query": query} ) # Parse the search results for content in response.content: results = json.loads(content.text) print(f"Found {len(results)} relevant tools for: {query}") for tool in results: print(f"- {tool['name']}: {tool['description']}") # Example usage await search_tools( "https://gateway-id.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp", "your-access-token", "How do I process images?" ) 

注文ID 123の情報は以下の通りです： - 注文ID: 123 - 状態: 処理中 (processing) - 商品: 1. 商品A: 2点 2. 商品B: 1点 - 合計金額: 5,000円 

boto3 botocore strands-agents bedrock-agentcore bedrock-agentcore-starter-toolkit python-dotenv requests 

# 仮想環境の作成と有効化 python3 -m venv agentcore-env source agentcore-env/bin/activate # 依存関係のインストール pip install -r requirements.txt 

. ├── requirements.txt # 依存関係 ├── .env # 環境変数設定（自動生成） ├── .env.example # 環境変数のテンプレート ├── setup_cognito.py # Cognito設定スクリプト（.env自動更新） ├── create_iam_roles.py # IAMロール作成スクリプト（.env自動更新） ├── create_lambda.py # Lambda関数作成スクリプト（.env自動更新） ├── create_gateway.py # Gateway作成スクリプト（.env自動更新） ├── setup_outbound_auth.py # OutboundAuth設定スクリプト（必須） ├── runtime_agent.py # Runtime実装（AgentCore Identity使用） ├── deploy_runtime.py # Runtimeデプロイスクリプト └── invoke_agentcore.py # 動作確認用スクリプト 

# Cognito設定（setup_cognito.pyが自動更新） COGNITO_DISCOVERY_URL= M2M_CLIENT_ID= M2M_CLIENT_SECRET= COGNITO_POOL_ID= RESOURCE_SERVER_ID=agentcore-gateway # Gateway設定（create_gateway.pyが自動更新） GATEWAY_URL= GATEWAY_ID= # IAMロール（create_iam_roles.pyが自動更新） GATEWAY_ROLE_ARN= RUNTIME_ROLE_ARN= LAMBDA_ROLE_ARN= # Lambda設定（create_lambda.pyが自動更新） LAMBDA_ARN= # AgentCore Identity設定（setup_outbound_auth.py実行後に手動追加） IDENTITY_PROVIDER_NAME=agentcore-identity-for-gateway GATEWAY_SCOPE=agentcore-gateway/read agentcore-gateway/write 

import boto3 import json import os from botocore.exceptions import ClientError from dotenv import load_dotenv, set_key REGION = "us-west-2" USER_POOL_NAME = "agentcore-gateway-pool" RESOURCE_SERVER_ID = "agentcore-gateway" RESOURCE_SERVER_NAME = "AgentCore Gateway Resource Server" CLIENT_NAME = "agentcore-m2m-client" # Scopeの定義 SCOPES = [ {"ScopeName": "read", "ScopeDescription": "Read access to Gateway"}, {"ScopeName": "write", "ScopeDescription": "Write access to Gateway"} ] def create_cognito_m2m_setup(): cognito = boto3.client("cognito-idp", region_name=REGION) try: # User Poolの作成 pool_response = cognito.create_user_pool( PoolName=USER_POOL_NAME, Policies={ 'PasswordPolicy': { 'MinimumLength': 8, 'RequireUppercase': True, 'RequireLowercase': True, 'RequireNumbers': True, 'RequireSymbols': True } } ) pool_id = pool_response['UserPool']['Id'] print(f"✅ User Pool作成完了: {pool_id}") # Resource Serverの作成 resource_response = cognito.create_resource_server( UserPoolId=pool_id, Identifier=RESOURCE_SERVER_ID, Name=RESOURCE_SERVER_NAME, Scopes=SCOPES ) print(f"✅ Resource Server作成完了: {RESOURCE_SERVER_ID}") # User Pool Domain作成（OAuth2 token endpointに必須） domain_name = f"agentcore-gateway-auth-{int(time.time())}" try: domain_response = cognito.create_user_pool_domain( Domain=domain_name, UserPoolId=pool_id ) print(f"✅ User Pool Domain作成完了: {domain_name}") except ClientError as e: if e.response['Error']['Code'] == 'InvalidParameterException' and 'already exists' in str(e): # ドメイン名重複時は別の名前で再試行 domain_name = f"agentcore-gateway-auth-{int(time.time())}-retry" domain_response = cognito.create_user_pool_domain( Domain=domain_name, UserPoolId=pool_id ) print(f"✅ User Pool Domain作成完了（リトライ）: {domain_name}") else: raise # App Client作成（M2M用） client_response = cognito.create_user_pool_client( UserPoolId=pool_id, ClientName=CLIENT_NAME, GenerateSecret=True, # M2M認証には必須 AllowedOAuthFlows=['client_credentials'], # M2M認証フロー AllowedOAuthScopes=[ f"{RESOURCE_SERVER_ID}/read", f"{RESOURCE_SERVER_ID}/write" ], AllowedOAuthFlowsUserPoolClient=True ) client_id = client_response['UserPoolClient']['ClientId'] client_secret = client_response['UserPoolClient']['ClientSecret'] # Discovery URLの生成 discovery_url = f"https://cognito-idp.{REGION}.amazonaws.com/{pool_id}/.well-known/openid-configuration" print("\n🎉 Cognito M2M設定完了！") print(f"Pool ID: {pool_id}") print(f"Client ID: {client_id}") print(f"Client Secret: {client_secret}") print(f"Discovery URL: {discovery_url}") # .envファイルに設定を自動保存 env_file = ".env" # .env.exampleからコピー（初回のみ） if not os.path.exists(env_file) and os.path.exists(".env.example"): with open(".env.example", "r") as src, open(env_file, "w") as dst: dst.write(src.read()) # 環境変数を更新 set_key(env_file, "COGNITO_POOL_ID", pool_id) set_key(env_file, "M2M_CLIENT_ID", client_id) set_key(env_file, "M2M_CLIENT_SECRET", client_secret) set_key(env_file, "COGNITO_DISCOVERY_URL", discovery_url) set_key(env_file, "RESOURCE_SERVER_ID", RESOURCE_SERVER_ID) print(f"\n✅ 設定を.envファイルに保存しました！") return { "pool_id": pool_id, "client_id": client_id, "client_secret": client_secret, "discovery_url": discovery_url } except ClientError as e: print(f"❌ エラーが発生しました: {e}") raise if __name__ == "__main__": cognito_config = create_cognito_m2m_setup() 

""" IAMロール作成スクリプト Gateway用とRuntime用のIAMロールを作成し、.envファイルに自動保存します。 """ import boto3 import json import os import time from botocore.exceptions import ClientError from dotenv import load_dotenv, set_key # AWSクライアント iam = boto3.client('iam') sts = boto3.client('sts') def get_account_id(): """AWSアカウントIDを取得""" return sts.get_caller_identity()["Account"] def create_trust_policy(account_id): """信頼ポリシーを生成（Gateway/Runtime共通）""" return { "Version": "2012-10-17", "Statement": [ { "Sid": "AssumeRolePolicy", "Effect": "Allow", "Principal": { "Service": "bedrock-agentcore.amazonaws.com" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "aws:SourceAccount": account_id }, "ArnLike": { "aws:SourceArn": f"arn:aws:bedrock-agentcore:us-west-2:{account_id}:*" } } } ] } 

# IAMロールを作成（.envファイルを自動更新） python create_iam_roles.py 

🚀 IAMロール作成スクリプトを開始します... アカウントID: 123456789012 📦 Gateway用IAMロールを作成中... ✅ IAMロール作成成功: AgentCoreGatewayRole-1234567890 ポリシーをアタッチしました 📦 Runtime用IAMロールを作成中... ✅ IAMロール作成成功: AgentCoreRuntimeRole-1234567890 ポリシーをアタッチしました 📦 Lambda関数用IAMロールを作成中... ✅ IAMロール作成成功: AgentCoreLambdaRole-1234567890 ポリシーをアタッチしました 🎉 IAMロール作成完了！ Gateway Role ARN: arn:aws:iam::123456789012:role/AgentCoreGatewayRole-1234567890 Runtime Role ARN: arn:aws:iam::123456789012:role/AgentCoreRuntimeRole-1234567890 Lambda Role ARN: arn:aws:iam::123456789012:role/AgentCoreLambdaRole-1234567890 ✅ .envファイルに保存しました 

import json def lambda_handler(event, context): """ AgentCore Gateway経由で呼び出されるLambda関数 context.client_contextからツール名を判別して処理を分岐 """ # Gateway経由の場合、context.client_contextが設定される tool_name = None try: if hasattr(context, 'client_context') and context.client_context: # client_contextから直接ツール名を取得 tool_name = context.client_context.custom['bedrockAgentCoreToolName'] print(f"Original tool name from Gateway: {tool_name}") # Gateway Target プレフィックスを除去 delimiter = "___" if delimiter in tool_name: tool_name = tool_name[tool_name.index(delimiter) + len(delimiter):] print(f"Processed tool name: {tool_name}") print(f"Client context structure: {str(context.client_context)}") else: print("No client_context available - direct Lambda invocation") except (AttributeError, KeyError, TypeError) as e: print(f"Error accessing client_context: {e}") tool_name = None # ツール名に基づいて処理を分岐 if tool_name == 'get_order_tool': order_id = event.get('orderId', 'unknown') # 実際のビジネスロジックをここに実装 result = { "orderId": order_id, "status": "processing", "items": [ {"name": "商品A", "quantity": 2}, {"name": "商品B", "quantity": 1} ], "total": 5000 } return { "statusCode": 200, "body": json.dumps(result) } elif tool_name == 'update_order_tool': order_id = event.get('orderId', 'unknown') # 実際の更新処理をここに実装 result = { "orderId": order_id, "status": "updated", "message": f"Order {order_id} has been updated successfully" } return { "statusCode": 200, "body": json.dumps(result) } else: # ツール名が不明な場合 return { "statusCode": 400, "body": json.dumps({ "error": f"Unknown tool: {tool_name}" }) } 

# Lambda関数を作成（.envファイルを自動更新） python create_lambda.py 

🚀 Lambda関数作成スクリプトを開始します... ✅ Lambda関数のコードを作成しました ✅ デプロイメントパッケージを作成しました: lambda_function_code.zip ✅ Lambda関数作成成功: agentcore-order-tools 🎉 Lambda関数作成完了！ Function ARN: arn:aws:lambda:us-west-2:123456789012:function:agentcore-order-tools ✅ .envファイルに保存しました 🧹 作業ファイルを削除しました 

""" AgentCore Identity OutboundAuth自動セットアップスクリプト Gateway用のCredential Providerを自動作成します。 """ import boto3 import requests import json import os from typing import Dict, Any, Optional from botocore.exceptions import ClientError class OutboundAuthSetup: def __init__(self, region: str = "us-west-2"): self.region = region self.cognito_client = boto3.client("cognito-idp", region_name=region) self.gateway_client = boto3.client("bedrock-agentcore-control", region_name=region) self.identity_client = boto3.client("bedrock-agentcore-control", region_name=region) def get_gateway_info(self, gateway_name: str) -> Dict[str, Any]: """ 既存のGatewayから必要な情報を取得 """ try: # Gatewayを検索 response = self.gateway_client.list_gateways() gateway_info = None for gateway in response.get("gateways", []): if gateway.get("name") == gateway_name: gateway_info = gateway break if not gateway_info: raise Exception(f"Gateway '{gateway_name}' not found") # 詳細情報を取得 gateway_detail = self.gateway_client.get_gateway( gatewayIdentifier=gateway_info["gatewayId"] ) print(f"✅ Gateway情報取得成功: {gateway_name}") print(f" Gateway ID: {gateway_detail['gatewayId']}") print(f" Gateway URL: {gateway_detail['gatewayUrl']}") return gateway_detail except ClientError as e: print(f"❌ Gateway情報取得エラー: {e}") raise def get_cognito_discovery_url(self, gateway_detail: Dict[str, Any]) -> str: """ GatewayのCognito設定からDiscovery URLを取得 """ try: auth_config = gateway_detail.get("authorizerConfiguration", {}) custom_jwt = auth_config.get("customJWTAuthorizer", {}) discovery_url = custom_jwt.get("discoveryUrl") if not discovery_url: raise Exception("Gateway authorizerConfiguration にdiscoveryUrlが見つかりません") print(f"✅ Discovery URL取得成功: {discovery_url}") return discovery_url except Exception as e: print(f"❌ Discovery URL取得エラー: {e}") raise def get_cognito_client_info(self, discovery_url: str) -> Dict[str, str]: """ Discovery URLからUser Pool IDを抽出し、クライアント情報を取得 """ try: # Discovery URLからUser Pool IDを抽出 # 例: https://cognito-idp.us-west-2.amazonaws.com/us-west-2_XXXXXXX/.well-known/openid-configuration import re match = re.search(r'/([^/]+)/.well-known', discovery_url) if not match: raise Exception("Discovery URLからUser Pool IDを抽出できません") user_pool_id = match.group(1) print(f"✅ User Pool ID抽出成功: {user_pool_id}") # User Pool Clientsを取得 clients_response = self.cognito_client.list_user_pool_clients( UserPoolId=user_pool_id ) if not clients_response.get("UserPoolClients"): raise Exception("User Pool Clientが見つかりません") # 最初のクライアント（通常M2M用）を使用 client_info = clients_response["UserPoolClients"][0] client_id = client_info["ClientId"] # クライアント詳細を取得 client_detail = self.cognito_client.describe_user_pool_client( UserPoolId=user_pool_id, ClientId=client_id ) client_secret = client_detail["UserPoolClient"].get("ClientSecret") print(f"✅ Cognito Client情報取得成功:") print(f" Client ID: {client_id}") print(f" Client Secret: {'*' * 8}...{client_secret[-4:] if client_secret else 'なし'}") return { "client_id": client_id, "client_secret": client_secret, "user_pool_id": user_pool_id, "discovery_url": discovery_url } except ClientError as e: print(f"❌ Cognito Client情報取得エラー: {e}") raise def create_oauth2_credential_provider(self, provider_name: str, cognito_info: Dict[str, str]) -> Dict[str, Any]: """ OAuth2 Credential Providerを作成 """ try: create_request = { "name": provider_name, "description": f"Gateway M2M認証用Credential Provider - {provider_name}", "configurationOverrides": { "clientId": cognito_info["client_id"], "clientSecret": cognito_info["client_secret"], "discoveryUrl": cognito_info["discovery_url"], "grantType": "client_credentials" } } print(f"🔐 OAuth2 Credential Provider作成中: {provider_name}") response = self.identity_client.create_oauth2_credential_provider(**create_request) print(f"✅ OAuth2 Credential Provider作成成功!") print(f" Provider Name: {response['name']}") print(f" Provider ID: {response.get('credentialProviderId', 'N/A')}") return response except ClientError as e: if e.response['Error']['Code'] == 'ConflictException': print(f"⚠️ Credential Provider '{provider_name}' は既に存在します") # 既存のプロバイダ情報を取得 try: list_response = self.identity_client.list_oauth2_credential_providers() for provider in list_response.get("oauth2CredentialProviders", []): if provider["name"] == provider_name: print(f"✅ 既存のCredential Provider使用: {provider_name}") return provider except Exception as list_error: print(f"❌ 既存プロバイダ情報取得エラー: {list_error}") raise else: print(f"❌ OAuth2 Credential Provider作成エラー: {e}") raise 

# OutboundAuth設定（必須ステップ）- .envファイルから自動読み取り python setup_outbound_auth.py 

🚀 OutboundAuth自動設定開始 .envファイルから設定を読み込み中... Gateway ID: 1234567890abcdef Provider Name: agentcore-identity-for-gateway Region: us-west-2 -------------------------------------------------- ✅ Gateway情報取得成功: order-management-gateway Gateway URL: https://gateway-1234567890abcdef.mcp.us-west-2.amazonaws.com ✅ Discovery URL取得成功: https://cognito-idp.us-west-2.amazonaws.com/us-west-2_XXXXXXX/.well-known/openid-configuration ✅ User Pool ID抽出成功: us-west-2_XXXXXXX ✅ Cognito Client情報取得成功: Client ID: 1234567890abcdef Client Secret: ********...xyz123 🔐 OAuth2 Credential Provider作成中: agentcore-identity-for-gateway ✅ OAuth2 Credential Provider作成成功! Provider Name: agentcore-identity-for-gateway Provider ID: resource-provider-oauth-client -------------------------------------------------- 🎉 OutboundAuth設定完了! ✅ 以下の環境変数を.envファイルに自動保存しました: IDENTITY_PROVIDER_NAME=agentcore-identity-for-gateway GATEWAY_SCOPE=agentcore-gateway/read agentcore-gateway/write 

import boto3 import json import os from bedrock_agentcore_starter_toolkit.operations.gateway.client import GatewayClient from dotenv import load_dotenv, set_key def create_gateway_with_lambda(cognito_config, gateway_role_arn, lambda_arn): """ Cognito認証付きGatewayを作成し、Lambda関数をMCPツール化 """ # Gateway clientの初期化 gateway_client = GatewayClient(region_name="us-west-2") # 1. Gatewayの作成（Cognito認証設定） gateway_name = "order-management-gateway" print(f"🚀 Gateway '{gateway_name}' を作成中...") try: authorizer_config = { "customJWTAuthorizer": { "discoveryUrl": cognito_config["discovery_url"], "allowedClients": [cognito_config["client_id"]] } } gateway_response = gateway_client.create_mcp_gateway( name=gateway_name, role_arn=gateway_role_arn, authorizer_config=authorizer_config, enable_semantic_search=True # セマンティック検索も有効化！ ) # gateway_responseは辞書なので、キーでアクセス gateway_id = gateway_response['gateway_id'] if 'gateway_id' in gateway_response else gateway_response['gatewayId'] gateway_url = gateway_response['mcp_url'] if 'mcp_url' in gateway_response else gateway_response['mcpUrl'] except Exception as e: if "already exists" in str(e) or "ConflictException" in str(e): print(f"ℹ️ Gateway '{gateway_name}' は既に存在します。既存のGatewayを取得中...") # 既存のGatewayを取得 gateways = gateway_client.list_gateways() existing_gateway = None # 名前で検索 for gateway in gateways: if gateway.get('name') == gateway_name: existing_gateway = gateway break if existing_gateway: gateway_id = existing_gateway['gateway_id'] if 'gateway_id' in existing_gateway else existing_gateway['gatewayId'] gateway_url = existing_gateway['mcp_url'] if 'mcp_url' in existing_gateway else existing_gateway['mcpUrl'] else: print("❌ 既存のGatewayが見つかりませんでした") raise e else: raise e print(f"✅ Gateway作成完了！") print(f" Gateway ID: {gateway_id}") print(f" MCP URL: {gateway_url}") # 2. Lambda関数をターゲットとして登録 print("\n🔧 Lambda関数をMCPツールとして登録中...") # ツールスキーマの定義 tool_schemas = [ { "name": "get_order_tool", "description": "注文情報を取得します", "inputSchema": { "type": "object", "properties": { "orderId": { "type": "string", "description": "注文ID" } }, "required": ["orderId"] } }, { "name": "update_order_tool", "description": "注文情報を更新します", "inputSchema": { "type": "object", "properties": { "orderId": { "type": "string", "description": "注文ID" } }, "required": ["orderId"] } } ] # ターゲット設定 target_config = { "mcp": { "lambda": { "lambdaArn": lambda_arn, "toolSchema": { "inlinePayload": tool_schemas } } } } # 認証情報プロバイダー（Lambda呼び出しにはGatewayのIAMロールを使用） credential_config = [ { "credentialProviderType": "GATEWAY_IAM_ROLE" } ] bedrock_client = boto3.client('bedrock-agentcore-control', region_name='us-west-2') target_response = bedrock_client.create_gateway_target( gatewayIdentifier=gateway_id, name="order-lambda-target", description="注文管理Lambda関数", targetConfiguration=target_config, credentialProviderConfigurations=credential_config ) print(f"✅ Lambda MCPツール登録完了！") print(f" Target ID: {target_response['targetId']}") # .envファイルに結果を保存 set_key(".env", "GATEWAY_ID", gateway_id) set_key(".env", "GATEWAY_URL", gateway_url) print(f"\n✅ Gateway情報を.envファイルに保存しました！") return { "gateway_id": gateway_id, "gateway_url": gateway_url, "target_id": target_response['targetId'] } if __name__ == "__main__": # .envファイルから設定を読み込み load_dotenv() # Cognito設定を.envから取得 cognito_config = { "discovery_url": os.environ.get("COGNITO_DISCOVERY_URL"), "client_id": os.environ.get("M2M_CLIENT_ID"), "client_secret": os.environ.get("M2M_CLIENT_SECRET") } # IAMロールとLambda ARNも.envから取得 gateway_role_arn = os.environ.get("GATEWAY_ROLE_ARN") lambda_arn = os.environ.get("LAMBDA_ARN") # 必須項目のチェック if not all([cognito_config["discovery_url"], cognito_config["client_id"], cognito_config["client_secret"], gateway_role_arn, lambda_arn]): print("❌ .envファイルに必要な設定が不足しています") print(" 以下の項目を確認してください:") print(" - COGNITO_DISCOVERY_URL, M2M_CLIENT_ID, M2M_CLIENT_SECRET") print(" - GATEWAY_ROLE_ARN, LAMBDA_ARN") exit(1) gateway_info = create_gateway_with_lambda(cognito_config, gateway_role_arn, lambda_arn) 

import os from strands import Agent from strands.models import BedrockModel from strands.tools.mcp import MCPClient from mcp.client.streamable_http import streamablehttp_client from bedrock_agentcore.runtime import BedrockAgentCoreApp from bedrock_agentcore.identity.auth import requires_access_token from typing import Dict, Any # エージェントアプリケーションの初期化 app = BedrockAgentCoreApp() @app.entrypoint async def order_management_agent(payload: Dict[str, Any]): """ 注文管理エージェント（AgentCore Identity使用） """ print("📋 エージェント起動") print(f"受信したペイロード: {payload}") # AgentCore Identityを使用してGatewayにアクセス gateway_url = os.environ.get("GATEWAY_URL") provider_name = os.environ.get("IDENTITY_PROVIDER_NAME", "agentcore-identity-for-gateway") gateway_scope = os.environ.get("GATEWAY_SCOPE") # 例: "agentcore-gateway/read agentcore-gateway/write" @requires_access_token( provider_name="resource-provider-oauth-client-zckzu", scopes=gateway_scope.split() if gateway_scope else [], auth_flow="M2M", force_authentication=False, ) async def process_with_gateway(*, access_token: str) -> str: """ Gatewayへのアクセストークンを取得し、MCPクライアントで処理 """ print(f"✅ アクセストークン取得成功") # MCPクライアントの作成（AgentCore Identity認証トークン付き） def create_streamable_http_transport(): return streamablehttp_client( gateway_url, headers={"Authorization": f"Bearer {access_token}"} ) client = MCPClient(create_streamable_http_transport) print(f"✅ MCP Client初期化完了（AgentCore Identity認証）") try: with client: # ツールリストを取得 tools = client.list_tools_sync() print(f"🛠️ 利用可能なツール: {[tool.tool_name for tool in tools]}") # Bedrockモデルとエージェントの初期化 model = BedrockModel( model_id="anthropic.claude-3-5-haiku-20241022-v1:0", params={"max_tokens": 4096, "temperature": 0.7}, region="us-west-2" ) agent = Agent( model=model, tools=tools, system_prompt="あなたは注文管理システムのアシスタントです。注文の照会や更新を手伝います。" ) print("✅ エージェント初期化完了！") # ユーザー入力を処理 user_input = payload.get("prompt", "注文IDの123の情報を教えて") print(f"💬 ユーザー入力: {user_input}") # エージェントで処理（内部でGatewayのツールを呼び出す） response = agent(user_input) result = response.message['content'][0]['text'] print(f"🤖 エージェント応答: {result}") return result except Exception as e: print(f"❌ エージェント処理エラー: {e}") return f"エラーが発生しました: {str(e)}" try: # AgentCore Identityを使用してアクセストークンを取得し、処理を実行 return await process_with_gateway() except Exception as e: print(f"❌ 認証エラー: {e}") return f"認証に失敗しました: {str(e)}" if __name__ == "__main__": app.run() 

# 初期設定 cp .env.example .env # Cognito設定（.envファイルを自動更新） python setup_cognito.py # → .envファイルにCognito設定が自動保存される # IAMロール作成（.envファイルを自動更新） python create_iam_roles.py # → Gateway用とRuntime用のロールARNが.envファイルに自動保存される # Lambda関数作成（.envファイルを自動更新） python create_lambda.py # → Lambda関数が作成され、ARNが.envファイルに自動保存される # Gateway作成（.envファイルを自動更新） python create_gateway.py # → .envファイルにGateway URLが自動保存される # OutboundAuth設定（必須ステップ） python setup_outbound_auth.py # → .envから自動読み取り、AgentCore IdentityのCredential Provider作成、結果を.envに自動保存 # Runtimeデプロイ（.envファイルから設定を読み込み、Runtime ARNも.envに自動保存） python deploy_runtime.py 

# 動作確認スクリプトを実行 python invoke_agentcore.py 

注文ID 123の情報は以下の通りです： - 注文ID: 123 - 状態: 処理中 (processing) - 商品: 1. 商品A: 2点 2. 商品B: 1点 - 合計金額: 5,000円 

{ "orderId": order_id, "status": "processing", "items": [ {"name": "商品A", "quantity": 2}, {"name": "商品B", "quantity": 1} ], "total": 5000 }