Obtenha o token de acesso OAuth 2.0

AgentCore O Identity permite que os desenvolvedores obtenham tokens OAuth para acesso delegado pelo usuário ou autenticação máquina a máquina com base nos provedores de credenciais OAuth 2.0 configurados. O serviço orquestrará o processo de autenticação entre o usuário ou o aplicativo até o servidor de autorização downstream e recuperará e armazenará o token resultante. Quando o token estiver disponível no AgentCore Identity Vault, agentes autorizados poderão recuperá-lo e usá-lo para autorizar chamadas para servidores de recursos. Por exemplo, o código de amostra abaixo recuperará um token para interagir com o Google Drive em nome de um usuário final. Para obter mais informações, consulte Integrar com o Google Drive usando o OAuth2 para ver o exemplo 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())

O processo é semelhante à obtenção de um token para chamadas de máquina a máquina, conforme mostrado no exemplo a seguir:

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

Armazenamento e uso de tokens de atualização automática

AgentCore armazena e usa automaticamente tokens de atualização quando disponíveis nos provedores do OAuth2, reduzindo a frequência das solicitações de reautorização do usuário. Quando os usuários concedem inicialmente o consentimento por meio de um fluxo de código de autorização padrão do OAuth2, o sistema armazena os tokens de acesso e os tokens de atualização (se fornecidos) no cofre seguro de tokens. Isso permite que os agentes obtenham novos tokens de acesso automaticamente quando os tokens originais expirarem, melhorando a experiência do usuário ao minimizar as repetidas solicitações de consentimento.

Os tokens de atualização geralmente têm uma vida útil mais longa do que os tokens de acesso, com um período de validade padrão de aproximadamente 30 dias em comparação com a vida útil mais curta dos tokens de acesso (geralmente de 1 a 2 horas). Quando um token de acesso expira, usa AgentCore automaticamente o token de atualização armazenado para solicitar um novo token de acesso ao provedor. Se um token de atualização válido for armazenado, AgentCore ignorará o fluxo de federação de usuários e retornará diretamente um novo token de acesso. Se o token de atualização também estiver expirado ou for inválido, o sistema voltará a solicitar ao usuário uma reautorização total.

Esse recurso não requer configuração interna AgentCore - ele opera automaticamente quando os tokens de atualização estão presentes na resposta do token do provedor OAuth2. No entanto, você deve configurar seu provedor OAuth2 para incluir tokens de atualização no fluxo de autorização. A configuração específica depende do seu provedor:

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

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

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

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

Se o seu provedor suportar tokens de atualização e estiver configurado corretamente, AgentCore ele os armazenará e gerenciará automaticamente sem configuração adicional. Para limpar os tokens de atualização armazenados e forçar os usuários a se autenticarem novamente, forceAuthentication=true defina ao ligar. GetResourceOauth2Token Isso limpa o token de atualização e força um fluxo de federação completo. Para obter informações sobre como configurar provedores do OAuth2, consulte Instalação e configuração do provedor.

URLs de autorização de streaming para chamadores de aplicativos

Para fluxos de OAuth (3LO) de três partes, seu agente precisa fornecer a URL de autorização ao aplicativo de chamada para que os usuários possam concluir o fluxo de consentimento. Embora os exemplos acima mostrem a impressão do URL no console, os aplicativos de produção precisam transmitir o URL de volta para o chamador por meio do mecanismo de resposta do seu aplicativo.

Padrões comuns de implementação

Padrão de resposta de streaming — Para aplicativos que oferecem suporte a respostas de streaming, você pode enviar o URL de autorização como parte do fluxo de resposta:

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

Padrão de retorno de chamada — Para aplicativos que usam retornos de chamada ou webhooks, armazene o URL de autorização e notifique o chamador:

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

Padrão de sondagem — Para aplicativos que preferem a sondagem, armazene o URL de autorização em um local recuperável:

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

Escolha o padrão que melhor se adapta à arquitetura do seu aplicativo. As respostas de streaming fornecem a melhor experiência do usuário para aplicativos em tempo real, enquanto os padrões de retorno de chamada e pesquisa funcionam bem para cenários de processamento assíncrono ou em lote.

Indicadores de recursos em fluxos do AgentCore OAuth2

Os indicadores de recursos fornecem uma forma padronizada de especificar qual servidor de recursos deve aceitar um token de acesso OAuth2. AgentCore usa o Cognito como seu provedor de autenticação, que oferece suporte a indicadores de recursos compatíveis com RFC 8707 que permitem especificar o servidor de recursos pretendido durante solicitações de token. Para usar indicadores de recursos, você deve primeiro configurar o servidor de autorização para reconhecer servidores de recursos específicos usando a CreateResourceServer API do Cognito. Depois de configurado, quando você especifica um indicador de recurso em sua solicitação de token, o Cognito inclui o identificador do servidor de recursos correspondente na declaração aud do token resultante, permitindo que o servidor de recursos verifique se o token se destina a seu uso específico. Isso fornece vários benefícios importantes: os servidores de recursos podem validar se os tokens são especificamente destinados a eles (princípio do menor privilégio), maior auditabilidade ao identificar claramente qual servidor de recursos cada token tem como alvo e redução do risco de uso indevido de tokens em diferentes serviços em seu ambiente de aplicativos.

Por meio da implementação do RFC 8707 do Cognito, AgentCore permite que os clientes especifiquem um servidor de recursos diretamente nas solicitações de autorização e token, substituindo o parâmetro de audiência padrão. No Cognito, o “indicador de recursos” referido no RFC corresponde ao valor do “identificador” doResourceServer. Os indicadores de recursos são particularmente importantes para implementações do Model Context Protocol (MCP), onde ajudam a mitigar riscos de segurança específicos descritos na especificação de autorização do MCP. O indicador de recursos corresponde ao parâmetro de recurso RFC 9728, garantindo o escopo adequado do token para interações com o servidor MCP. Observe que a implementação atual oferece suporte à vinculação de recurso único, o que significa que você pode especificar um servidor de recursos por solicitação de token.

Use indicadores de recursos quando seus agentes precisarem acessar servidores de recursos com requisitos de segurança específicos ou quando você precisar de um controle refinado sobre a validação do público de tokens. Os indicadores de recursos são particularmente úteis para aplicativos multilocatários em que os tokens devem ser restritos a recursos específicos do cliente.