OAuth 2.0 액세스 토큰 획득

AgentCore Identity를 사용하면 개발자가 구성된 OAuth 2.0 자격 증명 공급자를 기반으로 사용자 위임 액세스 또는 machine-to-machine 인증을 위한 OAuth 토큰을 얻을 수 있습니다. 서비스는 사용자 또는 애플리케이션 간에 인증 프로세스를 다운스트림 권한 부여 서버로 오케스트레이션하고 결과 토큰을 검색하여 저장합니다. AgentCore 자격 증명 볼트에서 토큰을 사용할 수 있게 되면 권한 있는 에이전트가 토큰을 검색하고 이를 사용하여 리소스 서버에 대한 호출을 승인할 수 있습니다. 예를 들어 아래 샘플 코드는 최종 사용자를 대신하여 Google Drive와 상호 작용할 토큰을 검색합니다. 자세한 내용은 전체 예제를 위해 OAuth2를 사용하여 Google Drive와 통합을 참조하세요.

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

이 프로세스는 다음 예제와 같이 machine-to-machine 호출에 대한 토큰을 얻는 것과 유사합니다.

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

자동 새로 고침 토큰 저장 및 사용

AgentCore는 OAuth2 공급자에서 사용 가능한 경우 새로 고침 토큰을 자동으로 저장하고 사용하므로 사용자 재인증 프롬프트의 빈도가 줄어듭니다. 사용자가 처음에 표준 OAuth2 권한 부여 코드 흐름을 통해 동의를 부여하면 시스템은 액세스 토큰과 새로 고침 토큰(제공된 경우)을 모두 보안 토큰 저장소에 저장합니다. 이를 통해 에이전트는 원래 토큰이 만료될 때 자동으로 새 액세스 토큰을 얻을 수 있으므로 반복되는 동의 요청을 최소화하여 사용자 경험을 개선할 수 있습니다.

새로 고침 토큰은 일반적으로 액세스 토큰보다 수명이 더 길며, 기본 유효 기간은 액세스 토큰의 더 짧은 수명(종종 1~2시간)에 비해 약 30일입니다. 액세스 토큰이 만료되면 AgentCore는 저장된 새로 고침 토큰을 자동으로 사용하여 공급자에게 새 액세스 토큰을 요청합니다. 유효한 새로 고침 토큰이 저장되면 AgentCore는 사용자 페더레이션 흐름을 건너뛰고 새 액세스 토큰을 직접 반환합니다. 새로 고침 토큰도 만료되거나 유효하지 않은 경우 시스템은 사용자에게 전체 재인증 메시지를 표시합니다.

이 기능은 AgentCore 내에서 구성이 필요하지 않습니다. 새로 고침 토큰이 OAuth2 공급자의 토큰 응답에 있으면 자동으로 작동합니다. 그러나 권한 부여 흐름에 새로 고침 토큰을 포함하도록 OAuth2 공급자를 구성해야 합니다. 특정 구성은 공급자에 따라 다릅니다.

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

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

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

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

공급자가 새로 고침 토큰을 지원하고 올바르게 구성된 경우 AgentCore는 추가 설정 없이 토큰을 자동으로 저장하고 관리합니다. 저장된 새로 고침 토큰을 지우고 사용자가 다시 인증하도록 하려면 GetResourceOauth2Token을 호출forceAuthentication=true할 때를 설정합니다. 이렇게 하면 새로 고침 토큰이 지워지고 전체 페더레이션 흐름이 강제로 실행됩니다. OAuth2 공급자 구성에 대한 자세한 내용은 공급자 설정 및 구성을 참조하세요.

애플리케이션 호출자에게 권한 부여 URLs 스트리밍

3각 OAuth(3LO) 흐름의 경우 에이전트는 사용자가 동의 흐름을 완료할 수 있도록 호출 애플리케이션에 권한 부여 URL을 제공해야 합니다. 위의 예제에서는 콘솔에 URL을 인쇄하는 것을 보여주지만 프로덕션 애플리케이션에서는 애플리케이션의 응답 메커니즘을 통해 호출자에게 URL을 다시 스트리밍해야 합니다.

일반적인 구현 패턴

스트리밍 응답 패턴 - 스트리밍 응답을 지원하는 애플리케이션의 경우 응답 스트림의 일부로 권한 부여 URL을 보낼 수 있습니다.

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

콜백 패턴 - 콜백 또는 웹후크를 사용하는 애플리케이션의 경우 권한 부여 URL을 저장하고 호출자에게 알립니다.

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

폴링 패턴 - 폴링을 선호하는 애플리케이션의 경우 권한 부여 URL을 검색 가능한 위치에 저장합니다.

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

애플리케이션 아키텍처에 가장 적합한 패턴을 선택합니다. 스트리밍 응답은 실시간 애플리케이션에 최상의 사용자 환경을 제공하는 반면, 콜백 및 폴링 패턴은 비동기 또는 배치 처리 시나리오에 적합합니다.

AgentCore OAuth2 흐름의 리소스 표시기

리소스 표시기는 OAuth2 액세스 토큰을 수락해야 하는 리소스 서버를 지정하는 표준화된 방법을 제공합니다. AgentCore는 Cognito를 인증 공급자로 사용하며, 이는 토큰 요청 중에 의도한 리소스 서버를 지정할 수 있는 RFC 8707 호환 리소스 표시기를 지원합니다. 리소스 표시기를 사용하려면 먼저 Cognito의 CreateResourceServer API를 사용하여 특정 리소스 서버를 인식하도록 권한 부여 서버를 구성해야 합니다. 구성이 완료되면 토큰 요청에 리소스 지표를 지정하면 Cognito는 결과 토큰의 aud 클레임에 해당 리소스 서버 식별자를 포함하여 리소스 서버가 토큰이 특정 용도인지 확인할 수 있습니다. 이렇게 하면 몇 가지 중요한 이점이 있습니다. 리소스 서버는 토큰이 전용인지 검증하고(최소 권한 원칙), 각 토큰이 대상으로 하는 리소스 서버를 명확하게 식별하여 감사 가능성을 개선하며, 애플리케이션 환경 내의 다양한 서비스에서 토큰 오용 위험을 줄일 수 있습니다.

Cognito의 RFC 8707 구현을 통해 AgentCore를 사용하면 클라이언트가 권한 부여 및 토큰 요청에서 직접 리소스 서버를 지정하여 기본 대상 파라미터를 재정의할 수 있습니다. Cognito에서 RFC에서 참조되는 '리소스 표시기'는 ResourceServer의 '식별자' 값에 해당합니다. 리소스 지표는 MCP 권한 부여 사양에 설명된 특정 보안 위험을 완화하는 데 도움이 되는 모델 컨텍스트 프로토콜(MCP) 구현에 특히 중요합니다. 리소스 표시기는 RFC 9728 리소스 파라미터에 대응하여 MCP 서버 상호 작용을 위한 적절한 토큰 크기 조정을 보장합니다. 현재 구현은 단일 리소스 바인딩을 지원하므로 토큰 요청당 하나의 리소스 서버를 지정할 수 있습니다.

에이전트가 특정 보안 요구 사항이 있는 리소스 서버에 액세스해야 하는 경우 또는 토큰 대상 검증을 세밀하게 제어해야 하는 경우 리소스 지표를 사용합니다. 리소스 표시기는 토큰을 특정 고객 리소스로 제한해야 하는 다중 테넌트 애플리케이션에 특히 유용합니다.