get_resource_oauth2_token async ¶

async def get_resource_oauth2_token(
    self,
    workload_identity_token: "capo_bedrock_agentcore.types.workload_identity_token_type.WorkloadIdentityTokenType",
    resource_credential_provider_name: "capo_bedrock_agentcore.types.credential_provider_name.CredentialProviderName",
    scopes: "capo_bedrock_agentcore.types.scopes_list_type.ScopesListType",
    oauth2_flow: "capo_bedrock_agentcore.types.oauth2_flow_type.Oauth2FlowType",
    *,
    config_overrides: Optional[AsyncBedrockAgentCoreClientConfig] = None,
    session_uri: Optional[
        "capo_bedrock_agentcore.types.request_uri.RequestUri"
    ] = None,
    resource_oauth2_return_url: Optional[
        "capo_bedrock_agentcore.types.resource_oauth2_return_url_type.ResourceOauth2ReturnUrlType"
    ] = None,
    force_authentication: Optional[bool] = None,
    custom_parameters: Optional[
        "capo_bedrock_agentcore.types.custom_request_parameters_type.CustomRequestParametersType"
    ] = None,
    custom_state: Optional["capo_bedrock_agentcore.types.state.State"] = None,
    resources: Optional[
        "capo_bedrock_agentcore.types.resources_list_type.ResourcesListType"
    ] = None,
    audiences: Optional[
        "capo_bedrock_agentcore.types.audiences_list_type.AudiencesListType"
    ] = None,
) -> "capo_bedrock_agentcore.types.get_resource_oauth2_token_response.GetResourceOauth2TokenResponse":
    """<p>Returns the OAuth 2.0 token of the provided resource.</p>

    Args:
        workload_identity_token: <p>The identity token of the workload from which you want to retrieve the OAuth2 token.</p>
        resource_credential_provider_name: <p>The name of the resource's credential provider.</p>
        scopes: <p>The OAuth scopes being requested.</p>
        oauth2_flow: <p>The type of flow to be performed.</p>
        session_uri: <p>Unique identifier for the user's authentication session for retrieving OAuth2 tokens. This ID tracks the authorization flow state across multiple requests and responses during the OAuth2 authentication process.</p>
        resource_oauth2_return_url: <p>The callback URL to redirect to after the OAuth 2.0 token retrieval is complete. This URL must be one of the provided URLs configured for the workload identity.</p>
        force_authentication: <p>Indicates whether to always initiate a new three-legged OAuth (3LO) flow, regardless of any existing session.</p>
        custom_parameters: <p>A map of custom parameters to include in the authorization request to the resource credential provider. These parameters are in addition to the standard OAuth 2.0 flow parameters, and will not override them.</p>
        custom_state: <p>An opaque string that will be sent back to the callback URL provided in resourceOauth2ReturnUrl. This state should be used to protect the callback URL of your application against CSRF attacks by ensuring the response corresponds to the original request.</p>
        resources: <p>The resources to include in the token request. These are used to specify the target resources for which the OAuth2 token is being requested.</p>
        audiences: <p>The audiences to include in the token request. These are used to specify the intended recipients of the OAuth2 token.</p>

    Raises:
        capo_bedrock_agentcore.errors.access_denied_exception.AccessDeniedException: <p>The exception that occurs when you do not have sufficient permissions to perform an action. Verify that your IAM policy includes the necessary permissions for the operation you are trying to perform.</p>
        capo_bedrock_agentcore.errors.internal_server_exception.InternalServerException: <p>The exception that occurs when the service encounters an unexpected internal error. This is a temporary condition that will resolve itself with retries. We recommend implementing exponential backoff retry logic in your application.</p>
        capo_bedrock_agentcore.errors.resource_not_found_exception.ResourceNotFoundException: <p>The exception that occurs when the specified resource does not exist. This can happen when using an invalid identifier or when trying to access a resource that has been deleted.</p>
        capo_bedrock_agentcore.errors.throttling_exception.ThrottlingException: <p>The exception that occurs when the request was denied due to request throttling. This happens when you exceed the allowed request rate for an operation. Reduce the frequency of requests or implement exponential backoff retry logic in your application.</p>
        capo_bedrock_agentcore.errors.unauthorized_exception.UnauthorizedException: <p>This exception is thrown when the JWT bearer token is invalid or not found for OAuth bearer token based access</p>
        capo_bedrock_agentcore.errors.validation_exception.ValidationException: <p>The exception that occurs when the input fails to satisfy the constraints specified by the service. Check the error message for details about which input parameter is invalid and correct your request.</p>
        capo_bedrock_agentcore.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_bedrock_agentcore.types.get_resource_oauth2_token_request.GetResourceOauth2TokenRequest]",
    ) -> AsyncOperationResponse[
        "capo_bedrock_agentcore.types.get_resource_oauth2_token_response.GetResourceOauth2TokenResponse"
    ]:
        import capo_bedrock_agentcore._operations.amazon_bedrock_agent_core.get_resource_oauth2_token

        (
            output,
            http_response,
        ) = await capo_bedrock_agentcore._operations.amazon_bedrock_agent_core.get_resource_oauth2_token.async_get_resource_oauth2_token(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_bedrock_agentcore.types.get_resource_oauth2_token_request.GetResourceOauth2TokenRequest",
    ) -> dict[str, Any]:
        import capo_bedrock_agentcore.types.get_resource_oauth2_token_request

        return capo_bedrock_agentcore.types.get_resource_oauth2_token_request.redact_sensitive(
            input_
        )

    def _redact_output(
        output: "capo_bedrock_agentcore.types.get_resource_oauth2_token_response.GetResourceOauth2TokenResponse",
    ) -> dict[str, Any]:
        import capo_bedrock_agentcore.types.get_resource_oauth2_token_response

        return capo_bedrock_agentcore.types.get_resource_oauth2_token_response.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "GetResourceOauth2Token",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_bedrock_agentcore.types.get_resource_oauth2_token_request.GetResourceOauth2TokenRequest = {
        "workload_identity_token": workload_identity_token,
        "resource_credential_provider_name": resource_credential_provider_name,
        "scopes": scopes,
        "oauth2_flow": oauth2_flow,
    }
    if session_uri is not None:
        input_["session_uri"] = session_uri
    if resource_oauth2_return_url is not None:
        input_["resource_oauth2_return_url"] = resource_oauth2_return_url
    if force_authentication is not None:
        input_["force_authentication"] = force_authentication
    if custom_parameters is not None:
        input_["custom_parameters"] = custom_parameters
    if custom_state is not None:
        input_["custom_state"] = custom_state
    if resources is not None:
        input_["resources"] = resources
    if audiences is not None:
        input_["audiences"] = audiences

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output