Obtenir un jeton d'accès OAuth 2.0

AgentCore L'identité permet aux développeurs d'obtenir des jetons OAuth pour un accès délégué par l'utilisateur ou une authentification machine à machine sur la base des fournisseurs d'informations d'identification OAuth 2.0 configurés. Le service orchestrera le processus d'authentification entre l'utilisateur ou l'application et le serveur d'autorisation en aval, et il récupérera et stockera le jeton obtenu. Une fois le jeton disponible dans le coffre AgentCore d'identité, les agents autorisés peuvent le récupérer et l'utiliser pour autoriser les appels aux serveurs de ressources. Par exemple, l'exemple de code ci-dessous permet de récupérer un jeton permettant d'interagir avec Google Drive pour le compte d'un utilisateur final. Pour plus d'informations, voir Intégrer à Google Drive à l'aide d'OAuth2 pour obtenir un exemple complet.

# 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,
    callback_url='insert_oauth2_callback_url_for_session_binding',
)
async def write_to_google_drive(*, access_token: str):
    # Use the token to call Google Drive
    pass

# To invoke:
# asyncio.run(write_to_google_drive())

Le processus est similaire à l'obtention d'un jeton pour les appels de machine à machine, comme illustré dans l'exemple suivant :

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
    pass

# To invoke:
# asyncio.run(need_token_2LO_async())

Stockage et utilisation des jetons d'actualisation automatiques

AgentCore stocke et utilise automatiquement les jetons d'actualisation lorsqu'ils sont disponibles auprès des fournisseurs OAuth2, réduisant ainsi la fréquence des demandes de réautorisation des utilisateurs. Lorsque les utilisateurs accordent initialement leur consentement par le biais d'un flux de code d'autorisation OAuth2 standard, le système stocke à la fois les jetons d'accès et les jetons d'actualisation (s'ils sont fournis) dans le coffre de jetons sécurisé. Cela permet aux agents d'obtenir automatiquement de nouveaux jetons d'accès lorsque les jetons d'origine expirent, améliorant ainsi l'expérience utilisateur en minimisant les demandes de consentement répétées.

Les jetons d'actualisation ont généralement une durée de vie plus longue que les jetons d'accès, avec une période de validité par défaut d'environ 30 jours par rapport à la durée de vie plus courte des jetons d'accès (souvent 1 à 2 heures). Lorsqu'un jeton d'accès expire, utilise AgentCore automatiquement le jeton d'actualisation stocké pour demander un nouveau jeton d'accès au fournisseur. Si un jeton d'actualisation valide est stocké, AgentCore ignore le flux de fédération d'utilisateurs et renvoie directement un nouveau jeton d'accès. Si le jeton d'actualisation est également expiré ou non valide, le système recommence à demander à l'utilisateur une réautorisation complète.

Cette fonctionnalité ne nécessite aucune configuration interne AgentCore  : elle fonctionne automatiquement lorsque des jetons d'actualisation sont présents dans la réponse des jetons du fournisseur OAuth2. Toutefois, vous devez configurer votre fournisseur OAuth2 pour inclure les jetons d'actualisation dans le flux d'autorisation. La configuration spécifique dépend de votre fournisseur :

"customParameters": { "access_type": "offline" }

"scopes": ["openid", "profile", "offline_access"]

"scopes": ["api", "refresh_token"]

"scopes": ["read:jira-user", "offline_access"]

Si votre fournisseur prend en charge les jetons d'actualisation et qu'il est correctement configuré, AgentCore il les stockera et les gérera automatiquement sans configuration supplémentaire. Pour effacer les jetons d'actualisation stockés et forcer les utilisateurs à s'authentifier à nouveau, configurez-les forceAuthentication=true lors de l'appel. GetResourceOauth2Token Cela efface le jeton d'actualisation et force un flux de fédération complet. Pour plus d'informations sur la configuration des fournisseurs OAuth2, consultez la section Configuration et configuration des fournisseurs.

URL d'autorisation de diffusion en continu destinées aux appelants de l'application

Pour les flux OAuth (3LO) à trois étapes, votre agent doit fournir l'URL d'autorisation à l'application appelante afin que les utilisateurs puissent terminer le flux de consentement. Bien que les exemples ci-dessus montrent l'impression de l'URL vers la console, les applications de production nécessitent de renvoyer l'URL à l'appelant via le mécanisme de réponse de votre application.

Modèles de mise en œuvre courants

Modèle de réponse en streaming : pour les applications qui prennent en charge les réponses en streaming, vous pouvez envoyer l'URL d'autorisation dans le cadre du flux de réponse :

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Stream URL back to caller instead of printing
    on_auth_url=lambda url: stream_to_caller({
        "type": "authorization_required",
        "authorization_url": url,
        "message": "Please visit this URL to authorize access"
    }),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_streaming_auth(*, access_token: str):
    # Agent logic continues after user completes authorization
    return {"status": "success", "token_received": True}

def stream_to_caller(data):
    # Implementation depends on your streaming mechanism
    # Examples: WebSocket, Server-Sent Events, HTTP chunked response
    response_stream.send(json.dumps(data))

Modèle de rappel : pour les applications utilisant des rappels ou des webhooks, stockez l'URL d'autorisation et informez l'appelant :

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Store URL and trigger callback
    on_auth_url=lambda url: handle_auth_callback(url),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_callback_auth(*, access_token: str):
    return {"status": "success", "data": "processed"}

def handle_auth_callback(authorization_url):
    # Store the URL associated with the request
    auth_store.save(request_id, {
        "authorization_url": authorization_url,
        "status": "pending_authorization"
    })

    # Notify the calling application
    callback_service.notify(callback_url, {
        "request_id": request_id,
        "authorization_url": authorization_url,
        "action_required": "user_authorization"
    })

Modèle de sondage : pour les applications qui préfèrent les sondages, stockez l'URL d'autorisation dans un emplacement récupérable :

import asyncio
from bedrock_agentcore.identity.auth import requires_access_token

@requires_access_token(
    provider_name="google-provider",
    scopes=["https://www.googleapis.com/auth/drive.metadata.readonly"],
    auth_flow="USER_FEDERATION",
    # Store URL for polling retrieval
    on_auth_url=lambda url: store_auth_url_for_polling(url),
    force_authentication=False,
    callback_url='insert_oauth2_callback_url_for_session_binding'
)
async def agent_with_polling_auth(*, access_token: str):
    return {"status": "success", "data": "processed"}

def store_auth_url_for_polling(authorization_url):
    # Store in database, cache, or session store
    session_store.set(f"auth_url:{session_id}", {
        "authorization_url": authorization_url,
        "created_at": datetime.utcnow(),
        "status": "pending"
    }, ttl=300)  # 5 minute expiration

Choisissez le modèle qui convient le mieux à l'architecture de votre application. Les réponses en streaming offrent la meilleure expérience utilisateur pour les applications en temps réel, tandis que les modèles de rappel et de sondage fonctionnent bien pour les scénarios de traitement asynchrone ou par lots.

Indicateurs de ressources dans les flux AgentCore OAuth2

Les indicateurs de ressources fournissent un moyen normalisé de spécifier quel serveur de ressources doit accepter un jeton d'accès OAuth2. AgentCore utilise Cognito comme fournisseur d'authentification, qui prend en charge les indicateurs de ressources conformes à la norme RFC 8707 qui vous permettent de spécifier le serveur de ressources prévu lors des demandes de jetons. Pour utiliser les indicateurs de ressources, vous devez d'abord configurer le serveur d'autorisation pour qu'il reconnaisse des serveurs de ressources spécifiques à l'aide de l' CreateResourceServer API de Cognito. Une fois configuré, lorsque vous spécifiez un indicateur de ressource dans votre demande de jeton, Cognito inclut l'identifiant du serveur de ressources correspondant dans la réclamation aud du jeton obtenu, ce qui permet au serveur de ressources de vérifier que le jeton est destiné à son usage spécifique. Cela présente plusieurs avantages importants : les serveurs de ressources peuvent vérifier que les jetons leur sont spécifiquement destinés (principe du moindre privilège), une meilleure auditabilité en identifiant clairement le serveur de ressources que chaque jeton cible et une réduction du risque d'utilisation abusive des jetons dans les différents services de votre environnement applicatif.

Grâce à l'implémentation de la RFC 8707 de Cognito, les clients AgentCore peuvent spécifier un serveur de ressources directement dans les demandes d'autorisation et de jeton, en remplaçant le paramètre d'audience par défaut. Dans Cognito, l' « indicateur de ressource » mentionné dans la RFC correspond à la valeur « identifiant » duResourceServer. Les indicateurs de ressources sont particulièrement importants pour les implémentations du protocole MCP (Model Context Protocol), où ils contribuent à atténuer les risques de sécurité spécifiques décrits dans la spécification d'autorisation MCP. L'indicateur de ressource correspond au paramètre de ressource de la RFC 9728, garantissant une définition appropriée des jetons pour les interactions avec le serveur MCP. Notez que l'implémentation actuelle prend en charge la liaison à ressource unique, ce qui signifie que vous pouvez spécifier un serveur de ressources par demande de jeton.

Utilisez des indicateurs de ressources lorsque vos agents ont besoin d'accéder à des serveurs de ressources soumis à des exigences de sécurité spécifiques, ou lorsque vous avez besoin d'un contrôle précis sur la validation des jetons d'audience. Les indicateurs de ressources sont particulièrement utiles pour les applications multi-locataires où les jetons doivent être limités à des ressources clients spécifiques.