最初の認証済みエージェントを構築する

この入門チュートリアルでは、Amazon Bedrock AgentCore Identity を使用して完全に認証されたエージェントをゼロから構築する手順について説明し、エージェントアプリケーションで ID 機能の実装を開始するのに役立ちます。開発環境をセットアップし、Cognito で認証インフラストラクチャを作成し、エージェントを AgentCore ランタイムにデプロイして、完全な認証ワークフローをテストする方法について説明します。

このチュートリアルを終了すると、OAuth2 フローを通じてユーザーを認証し、アクセストークンを安全に取得し、完全な ID 管理ライフサイクルを示すことができる、完全にデプロイされたエージェントが作成されます。エージェントは適切な IAM アクセス許可を持つ AgentCore ランタイムで実行され、統合機能をデモンストレーションしてテストできるテストラボ環境が作成されます。

前提条件

開始する前に、以下があることを確認してください。

このチュートリアルでは、OAuth 2.0 認可サーバーが必要です。持っていない場合は、ステップ 1 で Amazon Cognito ユーザープールを使用して作成します。クライアント ID、クライアントシークレット、およびユーザーが設定された OAuth 2.0 認可サーバーがある場合は、ステップ 2 に進むことができます。この認可サーバーはリソース認証情報プロバイダーとして機能し、エージェントにアウトバウンド OAuth 2.0 アクセストークンを付与する権限を表します。

SDK と依存関係をインストールする

このガイドのフォルダを作成し、Python 仮想環境を作成し、AgentCore SDK と Python SDK (boto3) AWS をインストールします。

mkdir agentcore-identity-quickstart
cd agentcore-identity-quickstart
python3 -m venv .venv
source .venv/bin/activate
pip install bedrock-agentcore boto3 strands-agents pyjwt

また、次の内容の requirements.txt ファイルを作成します。これは、後で AgentCore デプロイツールで使用されます。

bedrock-agentcore
boto3
pyjwt
strands-agents

ステップ 1: Cognito ユーザープールを作成する (オプション)

このチュートリアルでは、OAuth 2.0 認可サーバーが必要です。テストに使用できるものがない場合、またはテストを認可サーバーとは別に保持する場合、このスクリプトは AWS 認証情報を使用して Amazon Cognito インスタンスをセットアップし、認可サーバーとして使用します。スクリプトは以下を作成します。

Cognito ユーザープール AgentCoreIdentityQuickStartPool を削除すると、関連付けられた client_id とユーザーも削除されます。

このスクリプトを create_cognito.sh として保存してコマンドラインから実行するか、スクリプトをコマンドラインに貼り付けることができます。

#!/bin/bash

REGION=$(aws configure get region)

# Create user pool
USER_POOL_ID=$(aws cognito-idp create-user-pool \
  --pool-name AgentCoreIdentityQuickStartPool \
  --query 'UserPool.Id' \
  --no-cli-pager \
  --output text)

# Create user pool domain
DOMAIN_NAME="agentcore-quickstart-$(LC_ALL=C tr -dc 'a-z0-9' < /dev/urandom | head -c 5)"
aws cognito-idp create-user-pool-domain \
  --domain $DOMAIN_NAME \
  --no-cli-pager \
  --user-pool-id $USER_POOL_ID > /dev/null

# Create user pool client with secret and hosted UI settings
CLIENT_RESPONSE=$(aws cognito-idp create-user-pool-client \
  --user-pool-id $USER_POOL_ID \
  --client-name AgentCoreQuickStart \
  --generate-secret \
  --allowed-o-auth-flows "code" \
  --allowed-o-auth-scopes "openid" "profile" "email" \
  --allowed-o-auth-flows-user-pool-client \
  --supported-identity-providers "COGNITO" \
  --query 'UserPoolClient.{ClientId:ClientId,ClientSecret:ClientSecret}' \
  --output json)

CLIENT_ID=$(echo $CLIENT_RESPONSE | jq -r '.ClientId')
CLIENT_SECRET=$(echo $CLIENT_RESPONSE | jq -r '.ClientSecret')

# Generate random username and password
USERNAME="AgentCoreTestUser$(printf "%04d" $((RANDOM % 10000)))"
PASSWORD="$(LC_ALL=C tr -dc 'A-Za-z0-9!@#$%^&*()_+-=[]{}|;:,.<>?' < /dev/urandom | head -c 16)$(LC_ALL=C tr -dc '0-9' < /dev/urandom | head -c 1)"

# Create user with permanent password
aws cognito-idp admin-create-user \
  --user-pool-id $USER_POOL_ID \
  --username $USERNAME \
  --output text > /dev/null

aws cognito-idp admin-set-user-password \
  --user-pool-id $USER_POOL_ID \
  --username $USERNAME \
  --password $PASSWORD \
  --output text > /dev/null \
  --permanent

# Get region

ISSUER_URL="https://cognito-idp.$REGION.amazonaws.com/$USER_POOL_ID/.well-known/openid-configuration"
HOSTED_UI_URL="https://$DOMAIN_NAME.auth.$REGION.amazoncognito.com"

# Output results
echo "User Pool ID: $USER_POOL_ID"
echo "Client ID: $CLIENT_ID"
echo "Client Secret: $CLIENT_SECRET"
echo "Issuer URL: $ISSUER_URL"
echo "Hosted UI URL: $HOSTED_UI_URL"
echo "Test User: $USERNAME"
echo "Test Password: $PASSWORD"

echo ""
echo "# Copy and paste these exports to set environment variables for later use:"
echo "export USER_POOL_ID='$USER_POOL_ID'"
echo "export CLIENT_ID='$CLIENT_ID'"
echo "export CLIENT_SECRET='$CLIENT_SECRET'"
echo "export ISSUER_URL='$ISSUER_URL'"
echo "export HOSTED_UI_URL='$HOSTED_UI_URL'"
echo "export COGNITO_USERNAME='$USERNAME'"
echo "export COGNITO_PASSWORD='$PASSWORD'"

ステップ 2: 認証情報プロバイダーを作成する

認証情報プロバイダーは、エージェントが外部サービスにアクセスする方法です。認証情報プロバイダーを作成し、認可サーバーの OAuth 2.0 クライアントで設定します。

独自の認可サーバーを使用している場合は、環境変数 ISSUER_URL 、、および CLIENT_ID を認可サーバーからの適切な値CLIENT_SECRETで設定します。前のスクリプトを使用して Cognito で認可サーバーを作成する場合は、出力からターミナルに EXPORT ステートメントをコピーして環境変数を設定します。

この認証情報プロバイダーは、エージェントのコードによって使用され、ユーザーに代わって動作するアクセストークンを取得します。

ステップ 2.5: OAuth 2.0 認可サーバーにコールバック URL を追加する

不正なリダイレクトを防ぐには、CreateOauth2CredentialProvider または GetOauth2CredentialProvider から取得したコールバック URL を OAuth 2.0 認可サーバーに追加します。

前のスクリプトを使用して Cognito で認可サーバーを作成する場合は、出力からターミナルに EXPORT ステートメントをコピーして環境変数を設定し、Cognito ユーザープールクライアントを OAuth2 認証情報プロバイダーコールバック URL で更新します。

#!/bin/bash
aws cognito-idp update-user-pool-client \
    --user-pool-id $USER_POOL_ID \
    --client-id $CLIENT_ID \
    --client-name AgentCoreQuickStart \
    --allowed-o-auth-flows "code" \
    --allowed-o-auth-scopes "openid" "profile" "email" \
    --allowed-o-auth-flows-user-pool-client \
    --supported-identity-providers "COGNITO" \
    --callback-urls "$OAUTH2_CALLBACK_URL"

ステップ 3: OAuth 2.0 フローを開始するサンプルエージェントを作成する

このステップでは、OAuth 2.0 認可フローを開始し、ユーザーに代わって動作するトークンを取得するエージェントを作成します。わかりやすくするために、エージェントはユーザーに代わって外部サービスを実際に呼び出すことはありませんが、テストユーザーに代わって行動する同意を得たことを証明します。

エージェントコード

という名前のファイルを作成しagentcoreidentityquickstart.py、このコードを保存します。

"""
AgentCore Identity Outbound Token Agent

This agent demonstrates the USER_FEDERATION OAuth 2.0 flow.

It handles the OAuth 2.0 user consent flow and inspects the resulting OAuth 2.0 access token.
"""

from bedrock_agentcore.runtime import BedrockAgentCoreApp
from bedrock_agentcore.identity import requires_access_token
import asyncio
import jwt
import logging

app = BedrockAgentCoreApp()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def decode_jwt(token):
    try:
        decoded = jwt.decode(token, options={"verify_signature": False})
        return decoded
    except Exception as e:
        return {"error": f"Error decoding JWT: {str(e)}"}

class StreamingQueue:
    def __init__(self):
        self.finished = False
        self.queue = asyncio.Queue()

    async def put(self, item):
        await self.queue.put(item)

    async def finish(self):
        self.finished = True
        await self.queue.put(None)

    async def stream(self):
        while True:
            item = await self.queue.get()
            if item is None and self.finished:
                break
            yield item

queue = StreamingQueue()

async def handle_auth_url(url):
    await queue.put(f"Authorization URL, please copy to your preferred browser: {url}")

@requires_access_token(
    provider_name="AgentCoreIdentityQuickStartProvider",
    scopes=["openid"],
    auth_flow="USER_FEDERATION",
    on_auth_url=handle_auth_url, # streams authorization URL to client
    force_authentication=True,
    callback_url='insert_oauth2_callback_url_for_session_binding',
)
async def introspect_with_decorator(*, access_token: str):
    """Introspect token using decorator"""
    logger.info("Inside introspect_with_decorator - decorator succeeded")
    await queue.put({
        "message": "Successfully received an access token to act on behalf of your user!",
        "token_claims": decode_jwt(access_token),
        "token_length": len(access_token),
        "token_preview": f"{access_token[:50]}...{access_token[-10:]}"
    })
    await queue.finish()

@app.entrypoint
async def agent_invocation(payload, context):
    """Handler that uses only the decorator approach"""
    logger.info("Agent invocation started")

    # Start the agent task and immediately begin streaming
    task = asyncio.create_task(introspect_with_decorator())

    # Stream items as they come in
    async for item in queue.stream():
        yield item

    # Wait for task completion
    await task

if __name__ == "__main__":
    app.run()

ステップ 4: エージェントを AgentCore ランタイムにデプロイする

このエージェントは AgentCore ランタイムでホストされます。AgentCore CLI を使用すると、これを簡単に行うことができます。

ターミナルから AgentCore CLI をインストールし、新しいプロジェクトを作成します。

npm install -g @aws/agentcore
agentcore create --name IdentityQuickstart --defaults

エージェントスクリプトをプロジェクトのエージェントディレクトリにコピーし、デフォルトのエージェントを置き換えます。

cp agentcoreidentityquickstart.py IdentityQuickstart/app/IdentityQuickstart/main.py

また、要件ファイルをエージェントディレクトリにコピーして、すべての依存関係がデプロイに含まれていることを確認します。

cp requirements.txt IdentityQuickstart/app/IdentityQuickstart/

次に、プロジェクトをデプロイします。

cd IdentityQuickstart
agentcore deploy

CLI は AWS CDK スタックを合成し、エージェントを AgentCore ランタイムにデプロイします。これには約 2～3 分かかります。

トークンボールトとクライアントシークレットにアクセスできるようにエージェントの IAM ポリシーを更新する

AgentCore CLI はデプロイ中にエージェントの実行ロールを作成しますが、ロールにはトークンボールトアクセスのアクセス許可が自動的に含まれません。エージェントが実行時に OAuth 2.0 トークンを取得できるようにするには、追加のポリシーをアタッチする必要があります。

このスクリプトは、 CLI AWS からアカウントとリージョンを取得し、CloudFormation スタックからエージェントの実行ロールを検索して、適切なポリシーをアタッチします。このスクリプトをコピーして貼り付けるか、ファイルに保存して実行できます。

#!/bin/bash

# Get account and region from AWS CLI
AWS_ACCOUNT=$(aws sts get-caller-identity --query Account --output text)
REGION=$(aws configure get region)

# Get execution role from CloudFormation stack outputs
EXECUTION_ROLE=$(aws cloudformation describe-stack-resources \
  --stack-name AgentCore-IdentityQuickstart-prod \
  --query "StackResources[?ResourceType=='AWS::IAM::Role'].PhysicalResourceId" \
  --output text | head -1)

echo "Parsed values:"
echo "Execution Role: $EXECUTION_ROLE"
echo "Account: $AWS_ACCOUNT"
echo "Region: $REGION"

# Create the policy document with proper variable substitution
cat > agentcore-identity-policy.json << EOF
{
"Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AccessTokenVault",
      "Effect": "Allow",
      "Action": [
        "bedrock-agentcore:GetResourceOauth2Token",
        "secretsmanager:GetSecretValue"
      ],
      "Resource": ["arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:workload-identity-directory/default/workload-identity/*",
        "arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:token-vault/default/oauth2credentialprovider/AgentCoreIdentityQuickStartProvider",
        "arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:workload-identity-directory/default",
        "arn:aws:bedrock-agentcore:$REGION:$AWS_ACCOUNT:token-vault/default",
        "arn:aws:secretsmanager:$REGION:$AWS_ACCOUNT:secret:bedrock-agentcore-identity!default/oauth2/AgentCoreIdentityQuickStartProvider*"
      ]
    }
  ]
}
EOF

# Create the policy
POLICY_ARN=$(aws iam create-policy \
    --policy-name AgentCoreIdentityQuickStartPolicy$(LC_ALL=C tr -dc '0-9' < /dev/urandom | head -c 4) \
    --policy-document file://agentcore-identity-policy.json \
    --query 'Policy.Arn' \
    --output text)

# Extract role name from ARN and attach policy
ROLE_NAME=$(echo $EXECUTION_ROLE | awk -F'/' '{print $NF}')
aws iam attach-role-policy \
    --role-name $ROLE_NAME \
    --policy-arn $POLICY_ARN

echo "Policy created and attached: $POLICY_ARN"

# Cleanup
rm agentcore-identity-policy.json

ステップ 5: エージェントを呼び出す

これがすべて設定されたので、エージェントを呼び出すことができます。このデモでは、 agentcore invoke コマンドと IAM 認証情報を使用します。IAM 認証を使用するときは、 引数--user-idと --session-id引数を渡す必要があります。

agentcore invoke "TestPayload" --runtime IdentityQuickstart --user-id "SampleUserID" --session-id "ALongThirtyThreeCharacterMinimumSessionIdYouCanChangeThisAsYouNeed"

その後、エージェントはagentcore invokeコマンドへの URL を返します。その URL をコピーして任意のブラウザに貼り付けると、認可サーバーのログインページにリダイレクトされます。--user-id パラメータは、AgentCore Identity に提示するユーザー ID です。--session-id パラメータはセッション ID で、33 文字以上である必要があります。

ブラウザでプロンプトが表示されたら、認可サーバーでユーザーのユーザー名とパスワードを入力するか、設定した任意の認証方法を使用します。ステップ 1 のスクリプトを使用して Cognito インスタンスを作成した場合は、ターミナル履歴から取得できます。

ブラウザは、セッションバインディングフロー を処理する設定済みの OAuth2 コールバック URL にリダイレクトする必要があります。OAuth2 コールバックサーバーが、認可ステータスを示す明確な成功応答とエラー応答を提供することを確認します。

デバッグ

エラーや予期しない動作が発生した場合、エージェントの出力は Amazon CloudWatch logsにキャプチャされます。ログ末尾コマンドは、 の実行後に提供されますagentcore deploy。

クリーンアップ

完了したら、プロジェクトディレクトリagentcore deployから agentcore remove allと を実行して、デプロイされた AgentCore ランタイムリソースを破棄します。次に、Amazon Cognito ユーザープールを削除し、作成した IAM ポリシーをデタッチして削除し、認証情報プロバイダーを削除します。

セキュリティのベストプラクティス

ID 情報を使用する場合: