Obtain OAuth 2.0 access token - Amazon Bedrock AgentCore

Amazon Bedrock AgentCore is in preview release and is subject to change.

Obtain OAuth 2.0 access token

AgentCore Identity enables developers to obtain OAuth tokens for either user-delegated access or machine-to-machine authentication based on the configured OAuth 2.0 credential providers. The service will orchestrate the authentication process between the user or application to the downstream authorization server, and it will retrieve and store the resulting token. Once the token is available in the AgentCore Identity vault, authorized agents can retrieve it and use it to authorize calls to resource servers. For example, the sample code below will retrieve a token to interact with Google Drive on behalf of an end user. For more information, see Getting started with Amazon Bedrock AgentCore Identity for the complete example.

# 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, ) async def write_to_google_drive(*, access_token: str):     # Use the token to call Google Drive asyncio.run(write_to_google_drive(access_token=""))

The process is similar to obtain a token for machine-to-machine calls, as shown in the following example:

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 asyncio.run(need_token_2LO_async(access_token=""))