はじめに
全国1億2千万の EntraID 戦士のみなさんこんにちは、はしもと（仮名）です。
本文が長いのでさっさと本題に入ります。
概要
Amazon Bedrock AgentCore Identityを使用し、AgentCore RuntimeでホストしたAIエージェントの入出力の認証を実装します。
認証プロバイダーにはAWSで提供されるユーザー認証・認可サービス「Amazon Cognito」も利用できますが、今回はインバウンド認証に Microsoft Entra ID、アウトバウンド認証（リソースプロバイダ）にAtlassian Cloudを使用します。
ざっくりとした流れは、この赤色塗りつぶした経路を辿ります。
インバウンド認証とアウトバウンド認証とでは「アクセストークンを取得するタイミングをエージェントが判断するか」が異なり、前者はエージェントではなくクライアントから渡しますが、後者はエージェントがその判断をします。
これにより、以下のことが可能になります。
- 組織で Microsoft 365 サービスを利用するのと同じアカウントでエージェントの呼び出し可否を管理できる
- OAuthを使ってAtlassian Cloud側でもユーザーに認証を要求することで、ユーザーの持つアクセス範囲でエージェントを動かすことができる
Entra IDのユーザーとAtlassian Cloudでの権限が1対1にマッピングされているのではなく、それぞれで独立したユーザー認証を行うことに注意してください
ソースコード
手順
AWSリージョンは us-east-1（バージニア北部） リージョンを使用します。
またエージェントのデプロイや呼び出しには、Jupyter notebookの利用を推奨します。
Entraアプリケーションの準備
今回は簡単のため、デバイスフローを用いた認証を行います。
検証用のEntraテナントを用意とアプリの新規登録
名前は任意ですが、「サポートされているアカウントの種類」と「リダイレクトURI」は以下のようにしてください
- サポートされているアカウントの種類：任意の組織ディレクトリ内のアカウント (任意の Microsoft Entra ID テナント - マルチテナント)
- リダイレクトURI：Web - https://bedrock-agentcore.us-east-1.amazonaws.com/identities/oauth2/callback
API公開とカスタムスコープの追加
メニューの「APIの公開」から、公開用のURIを追加します。上部の「追加」を選択肢、そのまま「保存」を選択します。
続けて「Scopeの追加」 からカスタムスコープを追加します。これにより、のちに作成するAgentCore Runtimeで設定するallowedAudiencesとJWTのaudクレームの比較によりトークンの検証を組み込むことができ、このアプリ外で発行されたトークンを使ったリクエストを拒否することができます。
ここで作成するスコープ名はなんでも構いません。api://<アプリケーションID>/スコープ名で構成されるURLは控えておきます。
パブリッククライアントフローの許可
メニューの「認証」から「次のモバイルとデスクトップのフローを有効にする」をはいにします。
必要な認証情報の取得
アプリの概要ページから、以下の項目の値もコピーして控えておきます。
- アプリケーション（クライアント）ID
-
ディレクトリ（テナント）ID
Entra側に作成するアプリケーションの設定はこれで完了です。
Atlassian Cloudの準備
Atlassianアカウントの作成
検証に利用できるアカウントがない場合は、以下から無料のアカウントを作成してください。
Oauth2.0 連携用アプリケーション作成
Atlassian のデベロッパーコンソールから Create > OAuth 2.0 Integration を選択してアプリケーションを作成します。
qiita-atlassian-oauth という名前で作成しました
アプリケーション設定
Permissionの追加
作成したアプリに、Confluenceの操作の許可を追加します。メニューのPermissionsより、Confluence APIの行のAddを押すと、ボタンがConfigureに切り替わるのでもう一度選択します。
動作検証では、Confluenceページの取得と作成を行います。
Classic scopesのタブでsearch:confluence:、Granular scopesのタブでread:page:confluence、write:page:confluence、read:space:confluence、read:space-details:confluenceの計5つのスコープを追加します。
OAuth 2.0の設定
メニューのAuthorizationから認証設定を変更します。先ほど同様、OAuth 2.0 (3LO)の行のAddを選択し、遷移先の画面でCallback URLに https://bedrock-agentcore.us-east-1.amazonaws.com/identities/oauth2/callback を指定します。
クライアントIDとシークレットの取得
メニューのSettingsを選択すると、Authentication detailsのセクション内に記載されているので、これら2つの値を控えておきます。
以上で、Atlassian Cloudの設定も完了です。
OAuth クライアントを作成
AgentCore SDKを使用し、AgentCore Identityにアウトバウンド認証用 OAuth クライアントを作成していきます。
以下のコードを実行することで、AgentCoreの「アイデンティティ」内に、Atlassianをプロバイダーとする OAuth クライアントが作成されます。
OAuth クライアント作成
from bedrock_agentcore.services.identity import IdentityClient
from boto3.session import Session
boto_session = Session()
region = boto_session.region_name
identity_client = IdentityClient(region=region)
# Atlassian OAuth 2.0 アプリケーションの設定
os.environ["atlassian_client_id"] = "5SaT2NCocFhCgGW570hxqLhxLllPANgk"
os.environ["atlassian_secret"] = (
"ATOADnrNdM5owP4OB__CsrauPvF_YGw67ZTfRl5Lh-W5ZKvJdF8-PisYKgpbvtfD0vSC8373C33D"
)
os.environ["atlassian_scopes"] = (
"read:page:confluence write:page:confluence read:space:confluence offline_access"
)
# Provider（アイデンティティ）の作成
atlassian_provider = identity_client.create_oauth2_credential_provider(
req={
"name": "atlassian_oauth_provider",
"credentialProviderVendor": "AtlassianOauth2",
"oauth2ProviderConfigInput": {
"atlassianOauth2ProviderConfig": {
"clientId": os.environ["atlassian_client_id"],
"clientSecret": os.environ["atlassian_secret"],
}
},
}
)
（省略可）ローカルでの認証テスト
作成した OAuth クライアントを使ったアウトバウンド認証をローカルから試してみましょう。
アウトバウンド認証のテスト
from bedrock_agentcore.identity.auth import requires_access_token
@requires_access_token(
provider_name="atlassian_oauth_provider",
auth_flow="USER_FEDERATION",
scopes=os.environ["atlassian_scopes"].split(" "),
on_auth_url=lambda x: print(
"\n表示されたURLをブラウザのアドレスバーにコピーし認証を完了させてください:\n"
+ x
),
force_authentication=True,
)
def need_atlassian_access_token(*, access_token: str):
return access_token
need_atlassian_access_token()
表示されたURLを〜の下のURLをブラウザで開くと、Atlassian Cloudの認証画面に遷移するので、ログインした状態で Accept を選択します。
認証が完了しました！
notebook に戻ると、取得したアクセストークン（キャプチャ内eyJraで始まる文字列）が表示されています。
AgentCore Runtime のデプロイ
それでは、インバウンド認証に Microsoft Entra、アウトバウンド認証に Atlassian を設定したエージェントを AgentCore Runtime にデプロイします。
Runtime にデプロイするソースコードの詳細は、冒頭に記載の GitHub リポジトリ内 strands_confluence.py を参照ください。
Runtime 作成とデプロイ
agentcore_runtime = Runtime()
strands_e_a_response = agentcore_runtime.configure(
entrypoint="strands_confluence.py",
auto_create_execution_role=True,
auto_create_ecr=True,
requirements_file="requirements.txt",
region=region,
agent_name="strands_entra_confluence_3lo",
authorizer_configuration={
"customJWTAuthorizer": {
"discoveryUrl": discovery_url,
"allowedAudience": [os.environ["entra_audience"]],
} # ここでインバウンド認証の設定を追加
},
)
本記事のメイントピックであるインバウンド認証、アウトバウンド認証の設定箇所を確認します。
インバウンド認証設定
authorizer_configuration={
"customJWTAuthorizer": {
"discoveryUrl": discovery_url,
"allowedAudience": [os.environ["entra_audience"]],
} # ここでインバウンド認証の設定を追加
},
先ほど、Runtime 作成時に実行した agentcore_runtime.configure の中の authorizer_configuration に customJWTAuthorizer として Entra アプリの情報を指定しています。
Entra をインバウンド認証に使用する場合、JWTの検証には aud クレームを使用するため、許可する audience の一覧を allowedAudience として指定します
アウトバウンド認証設定
@requires_access_token(
provider_name="atlassian_oauth_provider",
scopes=os.environ.get("atlassian_scopes", "").split(),
auth_flow="USER_FEDERATION",
on_auth_url=on_auth_url,
force_authentication=False,
)
async def need_atlassian_token_async(*, access_token: str) -> str:
global atlassian_access_token, token_metadata
# トークン情報をデコードしてメタデータに保存
token_info = decode_token_info(access_token)
if token_info and token_info.get("exp") != "N/A":
exp_timestamp = token_info["exp"]
exp_datetime = datetime.fromtimestamp(exp_timestamp)
token_metadata["exp_time"] = exp_datetime.strftime("%Y-%m-%d %H:%M:%S")
atlassian_access_token = access_token
return access_token
アウトバウンド認証の設定はこの箇所です。Runtime にデプロイするソースコードの中に実装します。
ローカルから認証テストを行ったときと同様、@requires_access_token というデコレータを使います。このデコレータをつけたメソッドを他の関数から呼び出すことで、AgentCore Identity に作成したアウトバウンド認証用 OAuth クライアントへの認証とトークンの返却を要求します。
force_authentication=False というパラメータを設定しています。AgentCore Identity には token-vault という仕組みがあり、認証によって取得したアクセストークンやリフレッシュトークンをキャッシュのように保存することができます。
これにより、リフレッシュトークンの期間内であれば期限切れアクセストークンの再取得も AgentCore Identity がこっそりやってくれるため、Runtime はユーザーに認証を要求する頻度を減らすことができます。最高！
ユーザーに都度認証を求める場合はこのパラメータを True にします
https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/identity-authentication.html
インバウンド認証用アクセストークンの取得 (Entra)
ここでは、Microsoft の認証ライブラリである msal を使用します。
xxxxx となっている箇所は、準備段階で作成したご自身のアプリのクライアントID、スコープ、テナントIDに置き換えてください。
msal を使用した Entra 認証
import msal
import os
import webbrowser
# Entra ID の設定 (ユーザー認証用)
os.environ["entra_client_id"] = "xxxxx"
os.environ["entra_scopes"] = "api://xxxxx/qiita_test"
os.environ["entra_tenant_id"] = "xxxxx"
os.environ["entra_audience"] = "api://xxxxx"
REDIRECT_URI = (
f"https://bedrock-agentcore.{region}.amazonaws.com/identities/oauth2/callback"
)
AUTHORITY = f"https://login.microsoftonline.com/{os.environ['entra_tenant_id']}"
app = msal.PublicClientApplication(
client_id=os.environ["entra_client_id"],
authority=AUTHORITY,
)
entra_auth_result = app.acquire_token_silent(
scopes=[os.environ["entra_scopes"]], account=None
)
if not entra_auth_result:
flow = app.initiate_device_flow(scopes=[os.environ["entra_scopes"]])
if "user_code" not in flow:
raise ValueError("デバイスフローの開始に失敗しました。")
print(flow["message"])
webbrowser.open(flow["verification_uri"])
entra_auth_result = app.acquire_token_by_device_flow(flow)
access_token = entra_auth_result["access_token"]
print(f"Bearerトークンの取得完了: {access_token[:20]}...")
コードを実行するとブラウザで認証画面が開きます。出力された英数字の認証コードを入力します。
既にサインイン済みのアカウントがあれば下記の画面が表示されます。アカウント選択画面が表示された場合は、必ず テナントの管理者アカウント でサインインしてください。
アプリの要求スコープへの同意画面が表示されるので 承諾 を選択します。
認証が完了すると、Notebook の出力にアクセストークンの一部が出力されます。
このあとの動作確認でこのアクセストークンを使用するので、忘れずに認証を行ってください。
Runtime 経由で Confluence の API を呼び出す
大詰めです。AgentCore Runtime で実行しているAIエージェントを呼び出してみましょう！
実行の前に、Confluence へスペースを追加し、そのスペース内にテスト用のページを追加します。
私はこのように、「わっしょいスペース」というスペースと、「鳥貴族レポート」というページを作成しました。
ページ内容の取得
agentcore_runtime.invoke でリクエストを送信します。access_token には Entra 認証で取得した有効なアクセストークンを入れます。
prompt = "鳥貴族に関するページをConfluenceで検索し、内容を要約してください。"
result = agentcore_runtime.invoke(
payload={"prompt": prompt},
session_id=session_id,
bearer_token=access_token,
)
こちらが実行結果です。Authentication required ~ の部分は、AgentCore Identity が新しいトークンを要求していることを意味しています。
Runtime 上のエージェントを呼び出すのはこれが初めてなので、前述した taken-vault にはまだ利用可能なトークンが存在しないからですね。
新規ページの作成
次は、新しいページの作成をエージェントにお願いします。
prompt = "「わっしょいスペース」というスペースに、Amazon Bedrockに関する解説ページを新規作成してください。内容は200字程度でお願いします。"
result = agentcore_runtime.invoke(
payload={"prompt": prompt},
session_id=session_id,
bearer_token=access_token,
)
実行結果のログはこちらです。最初の実行とは異なり、認証がスキップされていることが分かります。
また、Confluence 上にもページが作成されていることが確認できます。
無効なアクセストークンを使用した場合（インバウンド認証失敗）
最後に、エラーとなるケースも見てみます。ここでは、Runtime へのリクエストに無効なアクセストークンを与え、インバウンド認証に失敗することを確認します。
prompt = "「わっしょいスペース」というスペースに、わっしょい祭り2025に関する解説ページを新規作成してください。内容は200字程度でお願いします。"
result = agentcore_runtime.invoke(
payload={"prompt": prompt},
session_id=session_id,
bearer_token="wasshoi_festival_2025", # 無効なトークンを指定
)
出力はこのようになります。HTTPError が発生し、Runtime の呼び出しに失敗したことが分かります。
重ねてになりますが、前述した customJWTAuthorizer でクレームの検証に失敗したため呼び出しの時点拒否される、という動きになります。
まとめ
AgentCore の主要機能である、Runtime と Identity を使用したインバウンド・アウトバウンド認証の検証を行いました。
検証にあたり、「プロバイダーにアプリを作成してスコープを追加して...」という作業はなかなか苦痛で時間がかかるものでしたが、そのおかげで OAuth/OIDC のしくみや AgentCore Identity がやってくれること/くれないことの理解がとても進んだと感じています。
話は飛びますが、普段利用している IaC ツールである Terraform にも、続々と AgentCore 対応の Pull Request がマージされています（日本人のコミッターの方が多くてただただ尊敬）。
次回検証の際は、Terraform でリソース管理を行い、手順も進めやすくできればと思っています。
はしもと（仮名）でした。
参考