OAuth 2.0 アクセストークンを取得する

AgentCore Identity を使用すると、デベロッパーは、設定された OAuth 2.0 認証情報プロバイダーに基づいて、ユーザー委任アクセスまたはmachine-to-machine認証のいずれかの OAuth トークンを取得できます。このサービスは、ユーザーまたはアプリケーション間の認証プロセスをダウンストリーム認可サーバーにオーケストレーションし、結果のトークンを取得して保存します。AgentCore Identity ボールトでトークンが利用可能になると、認可されたエージェントはトークンを取得し、それを使用してリソースサーバーへの呼び出しを承認できます。たとえば、以下のサンプルコードは、エンドユーザーに代わって Google Drive とやり取りするためのトークンを取得します。詳細については、完全な例のOAuth2 を使用した Google ドライブとの統合」を参照してください。

# Injects Google Access Token
@requires_access_token(
    # Uses the same credential provider name created above
    provider_name= "google-provider",
    # Requires Google OAuth2 scope to access Google Drive
    scopes= ["https://www.googleapis.com/auth/drive.metadata.readonly"],
    # Sets to OAuth 2.0 Authorization Code flow
    auth_flow= "USER_FEDERATION",
    # Prints authorization URL to console
    on_auth_url= lambda x: print("\nPlease copy and paste this URL in your browser:\n" + x),
    # If false, caches obtained access token
    force_authentication= False,
    callback_url='insert_oauth2_callback_url_for_session_binding',
)
async def write_to_google_drive(*, access_token: str):
    # Use the token to call Google Drive
    pass

# To invoke:
# asyncio.run(write_to_google_drive())

このプロセスは、次の例に示すように、machine-to-machine呼び出しのトークンを取得するのと似ています。

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token, requires_api_key

@requires_access_token(
    provider_name= "my-api-key-provider", # replace with your own credential provider name
    scopes= [],
    auth_flow= 'M2M',
)
async def need_token_2LO_async(*, access_token: str):
    # Use the access token
    pass

# To invoke:
# asyncio.run(need_token_2LO_async())

自動更新トークンのストレージと使用状況

AgentCore は、OAuth2 プロバイダーから利用可能な場合、更新トークンを自動的に保存して使用し、ユーザー再認可プロンプトの頻度を減らします。ユーザーが標準の OAuth2 認可コードフローを通じて最初に同意を付与すると、システムはアクセストークンと更新トークン (提供されている場合) の両方を安全なトークンボールトに保存します。これにより、エージェントは元のトークンの有効期限が切れると新しいアクセストークンを自動的に取得できるため、繰り返される同意リクエストを最小限に抑えることでユーザーエクスペリエンスが向上します。

更新トークンの有効期間は通常、アクセストークンよりも長く、デフォルトの有効期間は、アクセストークンの有効期間が短い (多くの場合 1～2 時間) と比較して約 30 日です。アクセストークンの有効期限が切れると、AgentCore は保存された更新トークンを自動的に使用して、プロバイダーに新しいアクセストークンをリクエストします。有効な更新トークンが保存されている場合、AgentCore はユーザーフェデレーションフローをスキップし、新しいアクセストークン を直接返します。更新トークンも有効期限切れまたは無効である場合、システムはユーザーに完全な再認可を求めるようにフォールバックします。

この機能は AgentCore 内で設定する必要はありません。OAuth2 プロバイダーのトークンレスポンスに更新トークンが存在すると自動的に動作します。ただし、認可フローに更新トークンを含めるように OAuth2 プロバイダーを設定する必要があります。特定の設定はプロバイダーによって異なります。

"customParameters": { "access_type": "offline" }

"scopes": ["openid", "profile", "offline_access"]

"scopes": ["api", "refresh_token"]

"scopes": ["read:jira-user", "offline_access"]

プロバイダーが更新トークンをサポートし、適切に設定されている場合、AgentCore は追加のセットアップなしでトークンを自動的に保存および管理します。保存された更新トークンをクリアし、ユーザーに再認証を強制するには、GetResourceOauth2Token を呼び出すforceAuthentication=trueときに を設定します。これにより、更新トークンがクリアされ、完全なフェデレーションフローが強制されます。OAuth2 プロバイダーの設定については、「プロバイダーのセットアップと設定」を参照してください。

アプリケーション発信者へのストリーミング承認 URLs

3 レッグの OAuth (3LO) フローの場合、ユーザーが同意フローを完了できるように、エージェントは呼び出し元のアプリケーションに認可 URL を指定する必要があります。上記の例は コンソールへの URL の印刷を示していますが、本番アプリケーションでは、アプリケーションの応答メカニズムを通じて URL を呼び出し元にストリーミングする必要があります。

一般的な実装パターン

ストリーミングレスポンスパターン – ストリーミングレスポンスをサポートするアプリケーションの場合、承認 URL をレスポンスストリームの一部として送信できます。

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Stream URL back to caller instead of printing
    on_auth_url=lambda url: stream_to_caller({
        "type": "authorization_required",
        "authorization_url": url,
        "message": "Please visit this URL to authorize access"
    }),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_streaming_auth(*, access_token: str):
    # Agent logic continues after user completes authorization
    return {"status": "success", "token_received": True}

def stream_to_caller(data):
    # Implementation depends on your streaming mechanism
    # Examples: WebSocket, Server-Sent Events, HTTP chunked response
    response_stream.send(json.dumps(data))

コールバックパターン – コールバックまたはウェブフックを使用するアプリケーションの場合は、認可 URL を保存し、発信者に通知します。

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Store URL and trigger callback
    on_auth_url=lambda url: handle_auth_callback(url),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_callback_auth(*, access_token: str):
    return {"status": "success", "data": "processed"}

def handle_auth_callback(authorization_url):
    # Store the URL associated with the request
    auth_store.save(request_id, {
        "authorization_url": authorization_url,
        "status": "pending_authorization"
    })

    # Notify the calling application
    callback_service.notify(callback_url, {
        "request_id": request_id,
        "authorization_url": authorization_url,
        "action_required": "user_authorization"
    })

ポーリングパターン – ポーリングを優先するアプリケーションの場合は、取得可能な場所に認可 URL を保存します。

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Store URL for polling retrieval
    on_auth_url=lambda url: store_auth_url_for_polling(url),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_polling_auth(*, access_token: str):
    return {"status": "success", "data": "processed"}

def store_auth_url_for_polling(authorization_url):
    # Store in database, cache, or session store
    session_store.set(f"auth_url:{session_id}", {
        "authorization_url": authorization_url,
        "created_at": datetime.utcnow(),
        "status": "pending"
    }, ttl=300)  # 5 minute expiration

アプリケーションアーキテクチャに最適なパターンを選択します。ストリーミングレスポンスはリアルタイムアプリケーションに最適なユーザーエクスペリエンスを提供し、コールバックパターンとポーリングパターンは非同期またはバッチ処理シナリオに適しています。

AgentCore OAuth2 フローのリソースインジケータ

リソースインジケータは、OAuth2 アクセストークンを受け入れるリソースサーバーを指定する標準化された方法を提供します。AgentCore は Cognito を認証プロバイダーとして使用します。これは、トークンリクエスト中に目的のリソースサーバーを指定できる RFC 8707 準拠のリソースインジケータをサポートします。リソースインジケータを使用するには、まず Cognito の CreateResourceServer API を使用して特定のリソースサーバーを認識するように認可サーバーを設定する必要があります。設定後、トークンリクエストでリソースインジケータを指定すると、Cognito は結果のトークンの aud クレームに対応するリソースサーバー識別子を含め、リソースサーバーがトークンが特定の使用を目的としていることを確認できるようにします。これはいくつかの重要な利点を提供します。リソースサーバーは、トークンがトークン専用であることを検証し (最小特権の原則）、各トークンターゲットを明確に識別することで監査可能性を向上させ、アプリケーション環境内のさまざまなサービスでトークンが悪用されるリスクを低減できます。

Cognito の RFC 8707 実装を通じて、AgentCore はクライアントが認可リクエストとトークンリクエストでリソースサーバーを直接指定し、デフォルトのオーディエンスパラメータを上書きできるようにします。Cognito では、RFC で参照される「リソースインジケータ」は ResourceServer の「識別子」値に対応します。リソース指標は、モデルコンテキストプロトコル (MCP) の実装にとって特に重要であり、MCP 認可仕様で説明されている特定のセキュリティリスクを軽減するのに役立ちます。リソースインジケータは RFC 9728 リソースパラメータに対応し、MCP サーバーインタラクションの適切なトークンスコープを確保します。現在の実装では、単一リソースバインディングがサポートされています。つまり、トークンリクエストごとに 1 つのリソースサーバーを指定できます。

エージェントが特定のセキュリティ要件を持つリソースサーバーにアクセスする必要がある場合、またはトークンオーディエンスの検証をきめ細かく制御する必要がある場合は、リソースインジケータを使用します。リソースインジケータは、トークンを特定の顧客リソースに制限する必要があるマルチテナントアプリケーションに特に役立ちます。