OAuth2 を使用して Google Drive と統合する

この入門チュートリアルでは、AI エージェントに Amazon Bedrock AgentCore Identity の使用を開始するための重要なステップについて説明します。開発環境のセットアップ、必要な SDKsのインストール、最初のエージェント ID の作成、エージェントが外部リソースに安全にアクセスできるようにする方法について説明します。

このチュートリアルを終了すると、AgentCore Identity OAuth2 認証情報プロバイダーを使用して Google からアクセストークンを取得し、アクセストークンを使用して Google Drive からファイルを読み取ることができる作業エージェントが作成されます。OAuth2 フローの詳細については、AgentCore Identity で認証情報プロバイダーを管理する」を参照してください。

前提条件

開始する前に、以下が必要です。

SDK のインストール

開始するには、 bedrock-agentcoreパッケージをインストールします。

pip install bedrock-agentcore

Google クライアント ID とクライアントシークレットを取得する

エージェントが Google Drive にアクセスできるようにするには、エージェントの Google クライアント ID とクライアントシークレットを取得する必要があります。Google デベロッパーコンソールに移動し、以下の手順に従います。

ステップ 1: OAuth 2.0 認証情報プロバイダーを設定する

次の CLI コマンドを使用して、以前に取得した Google クライアント ID とクライアントシークレットを使用して、新しい OAuth 2.0 AWS 認証情報プロバイダーを作成します。

OAUTH2_CREDENTIAL_PROVIDER_RESPONSE=$(aws bedrock-agentcore-control create-oauth2-credential-provider \
  --region us-east-1 \
  --name "google-provider" \
  --credential-provider-vendor "GoogleOauth2" \
  --oauth2-provider-config-input '{
      "googleOauth2ProviderConfig": {
        "clientId": "<your-google-client-id>",
        "clientSecret": "<your-google-client-secret>"
      }
    }' \
--output json)

OAUTH2_CALLBACK_URL=$(echo $OAUTH2_CREDENTIAL_PROVIDER_RESPONSE | jq -r '.callbackUrl')

echo "OAuth2 Callback URL: $OAUTH2_CALLBACK_URL"

ステップ 2: Identity モジュールと Auth モジュールをインポートする

このインポートステートメントを Python ファイルに追加します。

from bedrock_agentcore.services.identity import IdentityClient
from bedrock_agentcore.identity.auth import requires_access_token, requires_api_key

ステップ 3: OAuth 2.0 アクセストークンを取得する

前のステップで Google 認証情報プロバイダーを作成したら、Google アクセストークンを必要とするエージェントコードに@requires_access_tokenデコレータを追加します。コンソール出力から認可 URL をコピーし、ブラウザに貼り付けて、Google Drive で同意フローを完了します。

次のコードサンプルは、認可ワークフローを呼び出すためにエージェントコードに統合されることを目的としています。これは、個別にコピーして実行できるスタンドアロンコードではありません。

import asyncio

# 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,
    # The callback URL to redirect to after the OAuth 2.0 token retrieval is complete
    callback_url='oauth2_callback_url_for_session_binding',
)
async def write_to_google_drive(*, access_token: str):
    # Prints the access token obtained from Google
    print(access_token)

asyncio.run(write_to_google_drive(access_token=""))

バックグラウンドでは、@requires_access_tokenデコレータは次のシーケンスを実行します。

ステップ 4: OAuth2 アクセストークンを使用して外部リソースを呼び出す

エージェントが上記のステップで Google アクセストークンを取得すると、アクセストークンを使用して Google Drive にアクセスできます。ユーザーがアクセスできる最初の 10 個のファイルの名前と IDs を一覧表示する完全な例を次に示します。

まず、Python 用の Google クライアントライブラリをインストールします。

pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib

次に、次のコードをコピーします。

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token, requires_api_key
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

SCOPES = ["https://www.googleapis.com/auth/drive.metadata.readonly"]

def main(access_token):
    """Shows basic usage of the Drive v3 API.

    Prints the names and ids of the first 10 files the user has access to.
    """
    creds = Credentials(token=access_token, scopes=SCOPES)

    try:
        service = build("drive", "v3", credentials=creds)

        # Call the Drive v3 API
        results = (
            service.files()
            .list(pageSize=10, fields="nextPageToken, files(id, name)")
            .execute()
        )
        items = results.get("files", [])

        if not items:
            print("No files found.")
            return

        print("Files:")
        for item in items:
            print(f"{item['name']} ({item['id']})")

    except HttpError as error:
        # TODO(developer) - Handle errors from drive API.
        print(f"An error occurred: {error}")

if __name__ == "__main__":
    # This annotation helps agent developer to obtain access tokens from external applications
    @requires_access_token(
        provider_name="google-provider",
        # Google OAuth2 scopes
        scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
        # 3LO flow
        auth_flow="USER_FEDERATION",
        # prints authorization URL to console
        on_auth_url=lambda x: print("Copy and paste this authorization url to your browser", x),
        force_authentication=True,
        callback_url='oauth2_callback_url_for_session_binding',
    )
    async def read_from_google_drive(*, access_token: str):
        print(access_token)  # You can see the access_token
        # Make API calls...
        main(access_token)

    asyncio.run(read_from_google_drive(access_token=""))

次のステップ

このセクションの例では、特定のユースケースに適応できる実用的な実装パターンに焦点を当てています。エージェントまたはモデルコンテキストプロトコル (MCP) ツールの一部としてコードを埋め込むことができます。AgentCore Runtime でエージェントコードまたは MCP ツールをホストする場合は、Amazon Bedrock AgentCore Runtime でホストエージェントまたはツールを使用して、上記のコードを AgentCore Runtime にコピーします。