Intégrelo con Google Drive mediante OAuth2

En este tutorial de introducción, se explican los pasos esenciales para empezar a utilizar Amazon Bedrock AgentCore Identity para sus agentes de IA. Aprenderá a configurar su entorno de desarrollo, instalar los SDK necesarios, crear su primera identidad de agente y permitir que su agente acceda a recursos externos de forma segura.

Al final de este tutorial, dispondrá de un agente activo que podrá recuperar los tokens de acceso de Google con el proveedor de credenciales AgentCore Identity OAuth2 y leer archivos de Google Drive con los tokens de acceso. Para obtener información detallada sobre los flujos de OAuth2, consulta Administrar proveedores de credenciales con Identity. AgentCore

Requisitos previos

Antes de empezar, necesita:

Instalar el SDK

Para empezar, instala el bedrock-agentcore paquete:

pip install bedrock-agentcore

Obtenga el ID de cliente y el secreto de cliente de Google

Para que tu agente pueda acceder a Google Drive, debes obtener un ID de cliente de Google y un secreto de cliente para tu agente. Ve a la consola de desarrolladores de Google y sigue estos pasos:

Paso 1: Configura un proveedor de credenciales de OAuth 2.0

Crea un nuevo proveedor de credenciales de OAuth 2.0 con el ID de cliente de Google y el secreto de cliente obtenidos anteriormente mediante el siguiente comando CLI 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"

Paso 2: Importar los módulos de identidad y autenticación

Añada esta declaración de importación a su archivo de Python:

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

Paso 3: Obtén un token de acceso a OAuth 2.0

Una vez que hayas creado el proveedor de credenciales de Google en el paso anterior, añade el @requires_access_token decorador al código de agente que requiere un token de acceso de Google. Copia la URL de autorización de la salida de la consola, pégala en el navegador y completa el proceso de consentimiento con Google Drive.

El siguiente ejemplo de código está diseñado para integrarse en el código de su agente para invocar un flujo de trabajo de autorización. No se trata de un código independiente que se pueda copiar y ejecutar de forma independiente.

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

Entre bastidores, el @requires_access_token decorador sigue la siguiente secuencia:

Paso 4: Usa el token de acceso OAuth2 para invocar un recurso externo

Una vez que el agente obtenga un token de acceso de Google siguiendo los pasos anteriores, podrá usar el token de acceso para acceder a Google Drive. Este es un ejemplo completo que muestra los nombres e ID de los primeros 10 archivos a los que el usuario tiene acceso.

Primero, instala la biblioteca cliente de Google para Python:

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

A continuación, copia el siguiente código:

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

Pasos siguientes

El ejemplo de esta sección se centra en los patrones de implementación prácticos que puede adaptar a sus casos de uso específicos. Puede incrustar el código como parte de un agente o de una herramienta de protocolo de contexto modelo (MCP). Si desea alojar su código de agente o la herramienta MCP con AgentCore Runtime, siga el ejemplo del agente o las herramientas de alojamiento con Amazon Bedrock AgentCore Runtime para copiar el código anterior en Runtime. AgentCore