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 posterior, y recuperará y almacenará el token resultante. Una vez que el token esté disponible en la bóveda de AgentCore identidades, los agentes autorizados pueden recuperarlo y usarlo para autorizar las llamadas a los servidores de recursos. Por ejemplo, el código de ejemplo que aparece a continuación recuperará un token para interactuar con Google Drive en nombre de un usuario final. Para obtener más información, consulta Integrar con Google Drive mediante OAuth2 para ver el ejemplo completo.

# 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 usa 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 conceden su consentimiento inicialmente mediante un flujo de código 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 de tokens segura. Esto permite a los agentes obtener automáticamente nuevos tokens de acceso cuando los originales caducan, lo que mejora la experiencia del usuario al minimizar las solicitudes de consentimiento repetidas.

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 (a menudo de 1 a 2 horas). Cuando caduca un token de acceso, utiliza AgentCore automáticamente el token de actualización almacenado para solicitar al proveedor un nuevo token de acceso. 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, debes configurar tu proveedor de OAuth2 para incluir 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 administrará automáticamente sin necesidad de realizar ninguna configuración adicional. Para borrar los tokens de actualización almacenados y obligar a los usuarios a volver a autenticarse, configúralos forceAuthentication=true al llamar. GetResourceOauth2Token De este modo, se borra el token de actualización y se fuerza a un flujo de federación completo. Para obtener información sobre la configuración de los proveedores de OAuth2, consulte Instalación y configuración de proveedores.

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

En el caso de los flujos 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 cómo imprimir la URL en la consola, las aplicaciones de producción requieren que la URL se devuelva por streaming 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 en streaming: en el caso de las aplicaciones que admiten respuestas en streaming, puedes enviar la URL de autorización como parte del flujo de respuestas:

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, almacena 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: para las aplicaciones que prefieren el sondeo, almacena la URL de autorización en una ubicación que se pueda recuperar:

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 asincrónico 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 la RFC 8707, que permiten especificar el servidor de recursos deseado durante las solicitudes de tokens. Para usar 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 verificar que el token está destinado a un 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 de privilegio mínimo), mejorar la auditabilidad al identificar claramente a qué servidor de recursos se dirige cada token y reducir el riesgo de uso indebido de los tokens en los diferentes servicios del entorno de la aplicación.

Gracias a la implementación del RFC 8707 de Cognito, los clientes AgentCore 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 la RFC corresponde al valor del «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 de la RFC 9728, lo que garantiza que las interacciones entre los servidores MCP tengan el alcance adecuado de los tokens. Tenga en cuenta que la implementación actual admite la vinculación de un solo recurso, lo que significa que puede especificar un servidor de recursos por cada solicitud de token.

Usa los indicadores de recursos cuando tus agentes necesiten acceder a los servidores de recursos con requisitos de seguridad específicos o cuando necesites controlar con precisión la validación del público de los tokens. Los indicadores de recursos son especialmente útiles para las aplicaciones con varios usuarios, en las que los tokens deben restringirse a recursos específicos de los clientes.