Intégrer à Google Drive à l'aide d'OAuth2

Ce didacticiel de mise en route vous explique les étapes essentielles pour commencer à utiliser Amazon Bedrock AgentCore Identity pour vos agents d'intelligence artificielle. Vous apprendrez à configurer votre environnement de développement, à installer les SDK nécessaires, à créer votre première identité d'agent et à autoriser votre agent à accéder aux ressources externes en toute sécurité.

À la fin de ce didacticiel, vous disposerez d'un agent capable de récupérer des jetons d'accès auprès de Google avec le fournisseur d'informations d'identification AgentCore Identity OAuth2 et de lire des fichiers depuis Google Drive à l'aide de jetons d'accès. Pour des informations détaillées sur les flux OAuth2, voir Gérer les fournisseurs d'informations d'identification avec Identity. AgentCore

Conditions préalables

Avant de commencer, vous devez :

Installer le SDK

Pour commencer, installez le bedrock-agentcore package :

pip install bedrock-agentcore

Obtenir l'identifiant du client Google et le secret du client

Pour autoriser votre agent à accéder à Google Drive, vous devez obtenir un identifiant client Google et un secret client pour votre agent. Accédez à la console Google Developer et procédez comme suit :

Étape 1 : configurer un fournisseur d'informations d'identification OAuth 2.0

Créez un nouveau fournisseur d'informations d'identification OAuth 2.0 avec l'ID client Google et le secret client obtenus précédemment à l'aide de la commande CLI AWS suivante :

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"

Étape 2 : Importer les modules d'identité et d'authentification

Ajoutez cette instruction d'importation à votre fichier Python :

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

Étape 3 : Obtenir un jeton d'accès OAuth 2.0

Une fois que vous avez créé le fournisseur d'informations d'identification Google à l'étape précédente, ajoutez le @requires_access_token décorateur à votre code d'agent qui nécessite un jeton d'accès Google. Copiez l'URL d'autorisation depuis la sortie de votre console, puis collez-la dans votre navigateur et complétez le processus de consentement avec Google Drive.

L'exemple de code suivant est destiné à être intégré dans le code de votre agent pour appeler un flux de travail d'autorisation. Il ne s'agit pas d'un code autonome qui peut être copié et exécuté indépendamment.

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=""))

Dans les coulisses, le @requires_access_token décorateur suit la séquence suivante :

Étape 4 : utiliser le jeton d'accès OAuth2 pour appeler une ressource externe

Une fois que l'agent a obtenu un jeton d'accès Google en suivant les étapes ci-dessus, il peut utiliser le jeton d'accès pour accéder à Google Drive. Voici un exemple complet qui répertorie les noms et les identifiants des 10 premiers fichiers auxquels l'utilisateur a accès.

Tout d'abord, installez la bibliothèque cliente Google pour Python :

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

Copiez ensuite le code suivant :

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=""))

Quelle est la prochaine étape ?

L'exemple de cette section met l'accent sur des modèles de mise en œuvre pratiques que vous pouvez adapter à vos cas d'utilisation spécifiques. Vous pouvez intégrer le code dans un agent ou dans un outil MCP (Model Context Protocol). Si vous souhaitez héberger votre code d'agent ou votre outil MCP avec AgentCore Runtime, suivez l'agent hôte ou les outils avec Amazon Bedrock AgentCore Runtime pour copier le code ci-dessus dans Runtime. AgentCore