Obtenga el token de acceso de OAuth 2.0

AgentCore La identidad permite a los desarrolladores obtener tokens de OAuth para el acceso delegado por el usuario o para la autenticación de máquina a máquina en función de los proveedores de credenciales de OAuth 2.0 configurados. El servicio organizará el proceso de autenticación entre el usuario o la aplicación y el servidor de autorización derivado y recuperará y almacenará el token resultante. Una vez que el token esté disponible en el almacén de AgentCore identidades, los agentes autorizados podrán recuperarlo y usarlo para autorizar las llamadas a los servidores de recursos. Por ejemplo, en el siguiente código de ejemplo se recuperará un token para interactuar con Google Drive en nombre de un usuario final. Para obtener más información, consulta el ejemplo completo en Integrar con Google Drive mediante OAuth2.

# 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())

El proceso es similar al de obtener un token para las llamadas de máquina a máquina, como se muestra en el siguiente ejemplo:

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())

Almacenamiento y uso de los tokens de actualización automática

AgentCore almacena y utiliza automáticamente los tokens de actualización cuando están disponibles en los proveedores de OAuth2, lo que reduce la frecuencia de las solicitudes de reautorización de los usuarios. Cuando los usuarios dan su consentimiento inicialmente a través de un flujo de códigos de autorización estándar de OAuth2, el sistema almacena tanto los tokens de acceso como los de actualización (si se proporcionan) en la bóveda segura de tokens. Esto permite a los agentes obtener nuevos tokens de acceso automáticamente cuando los originales caducan, lo que mejora la experiencia del usuario al minimizar las solicitudes de consentimiento reiteradas.

Los tokens de actualización suelen tener una vida útil más larga que los de acceso, con un período de validez predeterminado de aproximadamente 30 días en comparación con la vida útil más corta de los tokens de acceso (normalmente de 1 a 2 horas). Cuando un token de acceso caduca, utiliza AgentCore automáticamente el token de actualización almacenado para solicitar un nuevo token de acceso al proveedor. Si se almacena un token de actualización válido, AgentCore omite el flujo de federación de usuarios y devuelve directamente un nuevo token de acceso. Si el token de actualización también ha caducado o no es válido, el sistema vuelve a solicitar al usuario la reautorización completa.

Esta función no requiere ninguna configuración interna AgentCore : funciona automáticamente cuando hay tokens de actualización en la respuesta del token del proveedor de OAuth2. Sin embargo, debe configurar su proveedor de OAuth2 para que incluya los tokens de actualización en el flujo de autorización. La configuración específica depende del proveedor:

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

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

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

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

Si su proveedor admite los tokens de actualización y está configurado correctamente, los AgentCore almacenará y gestionará automáticamente sin necesidad de realizar ninguna configuración adicional. Para borrar los identificadores de actualización almacenados y obligar a los usuarios a volver a autenticarse, configúrelos forceAuthentication=true al llamar. GetResourceOauth2Token Esto borra el token de actualización y fuerza un flujo de federación completo. Para obtener información sobre la configuración de los proveedores de OAuth2, consulte Configuración y configuración de proveedores.

Transmitir las direcciones URL de autorización a las personas que llaman a la aplicación

En el caso de los flujos de OAuth (3LO) de tres vías, el agente debe proporcionar la URL de autorización a la aplicación que realiza la llamada para que los usuarios puedan completar el flujo de consentimiento. Si bien en los ejemplos anteriores se muestra la impresión de la URL en la consola, las aplicaciones de producción requieren transmitir la URL a la persona que llama a través del mecanismo de respuesta de la aplicación.

Patrones de implementación comunes

Patrón de respuesta de transmisión: en el caso de las aplicaciones que admiten respuestas de transmisión, puede enviar la URL de autorización como parte del flujo de respuesta:

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

Patrón de devolución de llamada: en el caso de las aplicaciones que utilizan devoluciones de llamada o webhooks, guarda la URL de autorización y notifica a la persona que llama:

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

Patrón de sondeo: en el caso de las aplicaciones que prefieren el sondeo, guarde la URL de autorización en una ubicación recuperable:

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

Elija el patrón que mejor se adapte a la arquitectura de su aplicación. Las respuestas en streaming proporcionan la mejor experiencia de usuario para las aplicaciones en tiempo real, mientras que los patrones de devolución de llamadas y sondeos funcionan bien en escenarios de procesamiento asíncrono o por lotes.

Indicadores de recursos en los flujos de OAuth2 AgentCore

Los indicadores de recursos proporcionan una forma estandarizada de especificar qué servidor de recursos debe aceptar un token de acceso de OAuth2. AgentCore utiliza Cognito como proveedor de autenticación, que admite indicadores de recursos compatibles con RFC 8707 que permiten especificar el servidor de recursos previsto durante las solicitudes de token. Para utilizar los indicadores de recursos, primero debe configurar el servidor de autorización para que reconozca servidores de recursos específicos mediante la API de Cognito. CreateResourceServer Una vez configurado, al especificar un indicador de recursos en la solicitud de token, Cognito incluye el identificador del servidor de recursos correspondiente en la declaración aud del token resultante, lo que permite al servidor de recursos comprobar que el token está destinado a su uso específico. Esto proporciona varias ventajas importantes: los servidores de recursos pueden validar que los tokens están destinados específicamente a ellos (principio del mínimo privilegio), mejoran la auditabilidad al identificar claramente a qué servidor de recursos se dirige cada token y reducen el riesgo de uso indebido de los tokens en los diferentes servicios del entorno de aplicaciones.

Mediante la implementación del RFC 8707 de Cognito, AgentCore los clientes pueden especificar un servidor de recursos directamente en las solicitudes de autorización y de token, anulando el parámetro de audiencia predeterminado. En Cognito, el «indicador de recursos» al que se hace referencia en el RFC corresponde al valor de su «identificador». ResourceServer Los indicadores de recursos son particularmente importantes para las implementaciones del Protocolo de Contexto Modelo (MCP), ya que ayudan a mitigar los riesgos de seguridad específicos descritos en la especificación de autorización del MCP. El indicador de recursos corresponde al parámetro de recursos del RFC 9728, lo que garantiza una asignación adecuada de los símbolos a las interacciones con el servidor MCP. Tenga en cuenta que la implementación actual admite el enlace de un solo recurso, lo que significa que puede especificar un servidor de recursos por cada solicitud de token.

Utilice los indicadores de recursos cuando sus agentes necesiten acceder a servidores de recursos con requisitos de seguridad específicos o cuando necesite un control detallado de la validación de la audiencia de los tokens. Los indicadores de recursos son especialmente útiles para aplicaciones con varios usuarios, en las que los tokens deben restringirse a recursos específicos de los clientes.