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 e o servidor de autorização downstream e recuperará e armazenará o token resultante. Quando o token estiver disponível no AgentCore Identity Vault, os agentes autorizados poderão recuperá-lo e usá-lo para autorizar chamadas para servidores de recursos. Por exemplo, o código de exemplo 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 inicialmente concedem 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 expiram, melhorando a experiência do usuário ao minimizar as solicitações de consentimento repetidas.

Os tokens de atualização normalmente 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 do provedor. Se um token de atualização válido for armazenado, AgentCore ignora o fluxo de federação de usuários e retorna diretamente um novo token de acesso. Se o token de atualização também estiver expirado ou for inválido, o sistema volta a solicitar ao usuário uma reautorização completa.

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 seu provedor oferecer suporte a tokens de atualização e estiver configurado corretamente, os AgentCore 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 OAuth2, consulte Configuração e configuração do provedor.

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

Para fluxos OAuth (3LO) de três pernas, seu agente precisa fornecer o 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 exigem o streaming do 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 pesquisa — Para aplicativos que preferem a pesquisa, 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 oferecem a melhor experiência do usuário para aplicativos em tempo real, enquanto os padrões de retorno de chamada e sondagem funcionam bem em cenários de processamento assíncrono ou em lote.

Indicadores de recursos nos 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 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, primeiro você deve 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 oferece 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), melhor auditabilidade ao identificar claramente qual servidor de recursos cada token se destina e reduzir o 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 público padrão. No Cognito, o 'indicador de recursos' referido na RFC corresponde ao valor do 'identificador'. ResourceServer 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.