Step 4: Use OAuth2 Access Token to Invoke External Resource

Once the agent obtains a Google access token with the steps above, it can use the access token to access Google Drive. Here is a full example that lists the names and IDs of the first 10 files that the user has access to.

First, install the Google client library for Python:

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

Then, copy the following code:

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",
        scopes= ["https://www.googleapis.com/auth/drive.metadata.readonly"], # Google OAuth2 scopes
        auth_flow= "USER_FEDERATION", # 3LO flow
        on_auth_url= lambda x: print("Copy and paste this authorization url to your browser", x), # prints authorization URL to console
        force_authentication= True,)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= ""))