Obtenir un jeton d'accès OAuth 2.0

AgentCore Identity permet aux développeurs d'obtenir des jetons OAuth pour un accès délégué par l'utilisateur ou une authentification de 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, puis récupérera et stockera le jeton obtenu. Une fois que le jeton est disponible dans l' AgentCore Identity Vault, les agents autorisés peuvent le récupérer et l'utiliser pour autoriser les appels vers les 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 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 pour obtenir 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 des jetons d'actualisation lorsqu'ils sont disponibles auprès des fournisseurs OAuth2, ce qui réduit la fréquence des demandes de réautorisation des utilisateurs. Lorsque les utilisateurs accordent initialement leur consentement via un flux de code d'autorisation OAuth2 standard, le système stocke les jetons d'accès et les jetons d'actualisation (s'ils sont fournis) dans le coffre-fort à 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 a également expiré ou n'est pas valide, le système invite à nouveau l'utilisateur à effectuer une nouvelle 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 au jeton du fournisseur OAuth2. Cependant, vous devez configurer votre fournisseur OAuth2 pour inclure des 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 à se réauthentifier, 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, voir Configuration et configuration des fournisseurs.

Diffusion des URL d'autorisation vers les appelants de l'application

Pour les flux OAuth (3LO) à trois branches, votre agent doit fournir l'URL d'autorisation à l'application appelante afin que les utilisateurs puissent terminer le flux de consentement. Alors que les exemples ci-dessus montrent l'impression de l'URL vers la console, les applications de production nécessitent de retransmettre 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, enregistrez 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 le sondage, 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 d'interrogation 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 standardisé 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 RFC 8707 qui vous permettent de spécifier le serveur de ressources souhaité lors des demandes de jetons. Pour utiliser des 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, permettant au serveur de ressources de vérifier que le jeton est destiné à un usage spécifique. Cela présente plusieurs avantages importants : les serveurs de ressources peuvent valider que les jetons leur sont spécifiquement destinés (principe du moindre privilège), une auditabilité améliorée en identifiant clairement le serveur de ressources cible par chaque jeton 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 RFC 8707 de Cognito, elle AgentCore permet aux clients de spécifier un serveur de ressources directement dans les demandes d'autorisation et de jetons, 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 » ResourceServer de la RFC. 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 RFC 9728, garantissant une portée de jeton appropriée 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 doivent accéder à des serveurs de ressources présentant des exigences de sécurité spécifiques, ou lorsque vous avez besoin d'un contrôle précis de la validation de l'audience des jetons. 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.