start_async_invoke

Operation

start_async_invoke async

start_async_invoke(input: StartAsyncInvokeInput, plugins: list[Plugin] | None = None) -> StartAsyncInvokeOutput

Starts an asynchronous invocation.

This operation requires permission for the bedrock:InvokeModel action.

Warning

To deny all inference access to resources that you specify in the modelId field, you need to deny access to the bedrock:InvokeModel and bedrock:InvokeModelWithResponseStream actions. Doing this also denies access to the resource through the Converse API actions (Converse and ConverseStream). For more information see Deny access for inference on specific models.

Parameters:

Name	Type	Description	Default
input	StartAsyncInvokeInput	An instance of StartAsyncInvokeInput.	required
plugins	list[Plugin] | None	A list of callables that modify the configuration dynamically. Changes made by these plugins only apply for the duration of the operation execution and will not affect any other operation invocations.	None

Returns:

Type	Description
StartAsyncInvokeOutput	An instance of StartAsyncInvokeOutput.

Source code in src/aws_sdk_bedrock_runtime/client.py

async def start_async_invoke(
    self, input: StartAsyncInvokeInput, plugins: list[Plugin] | None = None
) -> StartAsyncInvokeOutput:
    """Starts an asynchronous invocation.

    This operation requires permission for the `bedrock:InvokeModel` action.

    Warning:
        To deny all inference access to resources that you specify in the
        modelId field, you need to deny access to the `bedrock:InvokeModel` and
        `bedrock:InvokeModelWithResponseStream` actions. Doing this also denies
        access to the resource through the Converse API actions
        ([Converse](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_Converse.html)
        and
        [ConverseStream](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ConverseStream.html)).
        For more information see [Deny access for inference on specific
        models](https://docs.aws.amazon.com/bedrock/latest/userguide/security_iam_id-based-policy-examples.html#security_iam_id-based-policy-examples-deny-inference).

    Args:
        input:
            An instance of `StartAsyncInvokeInput`.
        plugins:
            A list of callables that modify the configuration dynamically.
            Changes made by these plugins only apply for the duration of the
            operation execution and will not affect any other operation
            invocations.

    Returns:
        An instance of `StartAsyncInvokeOutput`.
    """
    operation_plugins: list[Plugin] = []
    if plugins:
        operation_plugins.extend(plugins)
    config = deepcopy(self._config)
    for plugin in operation_plugins:
        plugin(config)
    if config.protocol is None or config.transport is None:
        raise ExpectationNotMetError(
            "protocol and transport MUST be set on the config to make calls."
        )
    pipeline = RequestPipeline(protocol=config.protocol, transport=config.transport)
    call = ClientCall(
        input=input,
        operation=START_ASYNC_INVOKE,
        context=TypedProperties({"config": config}),
        interceptor=InterceptorChain(config.interceptors),
        auth_scheme_resolver=config.auth_scheme_resolver,
        supported_auth_schemes=config.auth_schemes,
        endpoint_resolver=config.endpoint_resolver,
        retry_strategy=config.retry_strategy,
    )

    return await pipeline(call)

Input

StartAsyncInvokeInput dataclass

Dataclass for StartAsyncInvokeInput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

@dataclass(kw_only=True)
class StartAsyncInvokeInput:
    """Dataclass for StartAsyncInvokeInput structure."""

    client_request_token: str | None = None
    """Specify idempotency token to ensure that requests are not duplicated."""

    model_id: str | None = None
    """The model to invoke."""

    model_input: Document | None = field(repr=False, default=None)
    """Input to send to the model."""

    output_data_config: AsyncInvokeOutputDataConfig | None = None
    """Where to store the output."""

    tags: list[Tag] | None = None
    """Tags to apply to the invocation."""

    def serialize(self, serializer: ShapeSerializer):
        serializer.write_struct(_SCHEMA_START_ASYNC_INVOKE_INPUT, self)

    def serialize_members(self, serializer: ShapeSerializer):
        if self.client_request_token is not None:
            serializer.write_string(
                _SCHEMA_START_ASYNC_INVOKE_INPUT.members["clientRequestToken"],
                self.client_request_token,
            )

        if self.model_id is not None:
            serializer.write_string(
                _SCHEMA_START_ASYNC_INVOKE_INPUT.members["modelId"], self.model_id
            )

        if self.model_input is not None:
            serializer.write_document(
                _SCHEMA_START_ASYNC_INVOKE_INPUT.members["modelInput"], self.model_input
            )

        if self.output_data_config is not None:
            serializer.write_struct(
                _SCHEMA_START_ASYNC_INVOKE_INPUT.members["outputDataConfig"],
                self.output_data_config,
            )

        if self.tags is not None:
            _serialize_tag_list(
                serializer, _SCHEMA_START_ASYNC_INVOKE_INPUT.members["tags"], self.tags
            )

    @classmethod
    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
        return cls(**cls.deserialize_kwargs(deserializer))

    @classmethod
    def deserialize_kwargs(cls, deserializer: ShapeDeserializer) -> dict[str, Any]:
        kwargs: dict[str, Any] = {}

        def _consumer(schema: Schema, de: ShapeDeserializer) -> None:
            match schema.expect_member_index():
                case 0:
                    kwargs["client_request_token"] = de.read_string(
                        _SCHEMA_START_ASYNC_INVOKE_INPUT.members["clientRequestToken"]
                    )

                case 1:
                    kwargs["model_id"] = de.read_string(
                        _SCHEMA_START_ASYNC_INVOKE_INPUT.members["modelId"]
                    )

                case 2:
                    kwargs["model_input"] = de.read_document(
                        _SCHEMA_START_ASYNC_INVOKE_INPUT.members["modelInput"]
                    )

                case 3:
                    kwargs["output_data_config"] = (
                        _AsyncInvokeOutputDataConfigDeserializer().deserialize(de)
                    )

                case 4:
                    kwargs["tags"] = _deserialize_tag_list(
                        de, _SCHEMA_START_ASYNC_INVOKE_INPUT.members["tags"]
                    )

                case _:
                    logger.debug("Unexpected member schema: %s", schema)

        deserializer.read_struct(_SCHEMA_START_ASYNC_INVOKE_INPUT, consumer=_consumer)
        return kwargs

Attributes

client_request_token class-attribute instance-attribute

client_request_token: str | None = None

Specify idempotency token to ensure that requests are not duplicated.

model_id class-attribute instance-attribute

model_id: str | None = None

The model to invoke.

model_input class-attribute instance-attribute

model_input: Document | None = field(repr=False, default=None)

Input to send to the model.

output_data_config class-attribute instance-attribute

output_data_config: AsyncInvokeOutputDataConfig | None = None

Where to store the output.

tags class-attribute instance-attribute

tags: list[Tag] | None = None

Tags to apply to the invocation.

Output

StartAsyncInvokeOutput dataclass

Dataclass for StartAsyncInvokeOutput structure.

Source code in src/aws_sdk_bedrock_runtime/models.py

@dataclass(kw_only=True)
class StartAsyncInvokeOutput:
    """Dataclass for StartAsyncInvokeOutput structure."""

    invocation_arn: str
    """The ARN of the invocation."""

    def serialize(self, serializer: ShapeSerializer):
        serializer.write_struct(_SCHEMA_START_ASYNC_INVOKE_OUTPUT, self)

    def serialize_members(self, serializer: ShapeSerializer):
        serializer.write_string(
            _SCHEMA_START_ASYNC_INVOKE_OUTPUT.members["invocationArn"],
            self.invocation_arn,
        )

    @classmethod
    def deserialize(cls, deserializer: ShapeDeserializer) -> Self:
        return cls(**cls.deserialize_kwargs(deserializer))

    @classmethod
    def deserialize_kwargs(cls, deserializer: ShapeDeserializer) -> dict[str, Any]:
        kwargs: dict[str, Any] = {}

        def _consumer(schema: Schema, de: ShapeDeserializer) -> None:
            match schema.expect_member_index():
                case 0:
                    kwargs["invocation_arn"] = de.read_string(
                        _SCHEMA_START_ASYNC_INVOKE_OUTPUT.members["invocationArn"]
                    )

                case _:
                    logger.debug("Unexpected member schema: %s", schema)

        deserializer.read_struct(_SCHEMA_START_ASYNC_INVOKE_OUTPUT, consumer=_consumer)
        return kwargs

Attributes

invocation_arn instance-attribute

invocation_arn: str

The ARN of the invocation.

Errors

• AccessDeniedException

• ConflictException

• InternalServerException

• ResourceNotFoundException

• ServiceQuotaExceededException

• ServiceUnavailableException

• ThrottlingException

• ValidationException