Invokes a model endpoint with bidirectional streaming capabilities. This operation establishes a persistent connection that allows you to send multiple requests and receive streaming responses from the model in real-time.
Bidirectional streaming is useful for interactive applications such as chatbots, real-time translation, or any scenario where you need to maintain a conversation-like interaction with the model. The connection remains open, allowing you to send additional input and receive responses without establishing a new connection for each request.
For an overview of Amazon SageMaker AI, see How It Works.
Amazon SageMaker AI strips all POST headers except those supported by the API. Amazon SageMaker AI might add additional headers. You should not rely on the behavior of headers outside those enumerated in the request syntax.
Calls to InvokeEndpointWithBidirectionalStream are authenticated by using Amazon Web Services Signature Version 4. For information, see Authenticating Requests (Amazon Web Services Signature Version 4) in the Amazon S3 API Reference.
The bidirectional stream maintains the connection until either the client closes it or the model indicates completion. Each request and response in the stream is sent as an event with optional headers for data type and completion state.
Endpoints are scoped to an individual account, and are not public. The URL does not contain the account ID, but Amazon SageMaker AI determines the account ID from the authentication token that is supplied by the caller.
Parameters:
Raises:
Source code in docs_src/capo_sagemaker_runtime_http2/_services/async_sage_maker_runtime_http2.py
| @asynccontextmanager
async def invoke_endpoint_with_bidirectional_stream(
self,
endpoint_name: str,
body: "AsyncIterator[capo_sagemaker_runtime_http2.types.request_stream_event._RequestStreamEvent] | capo_sagemaker_runtime_http2.types.request_stream_event._RequestStreamEvent",
*,
config_overrides: Optional[AsyncSageMakerRuntimeHTTP2ClientConfig] = None,
target_variant: Optional[str] = None,
model_invocation_path: Optional[str] = None,
model_query_string: Optional[str] = None,
) -> "AsyncGenerator[capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_output.InvokeEndpointWithBidirectionalStreamOutput]":
"""<p>Invokes a model endpoint with bidirectional streaming capabilities. This operation establishes a persistent connection that allows you to send multiple requests and receive streaming responses from the model in real-time.</p> <p>Bidirectional streaming is useful for interactive applications such as chatbots, real-time translation, or any scenario where you need to maintain a conversation-like interaction with the model. The connection remains open, allowing you to send additional input and receive responses without establishing a new connection for each request.</p> <p>For an overview of Amazon SageMaker AI, see <a href="https://docs.aws.amazon.com/sagemaker/latest/dg/how-it-works.html">How It Works</a>.</p> <p>Amazon SageMaker AI strips all POST headers except those supported by the API. Amazon SageMaker AI might add additional headers. You should not rely on the behavior of headers outside those enumerated in the request syntax. </p> <p>Calls to <code>InvokeEndpointWithBidirectionalStream</code> are authenticated by using Amazon Web Services Signature Version 4. For information, see <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/sig-v4-authenticating-requests.html">Authenticating Requests (Amazon Web Services Signature Version 4)</a> in the <i>Amazon S3 API Reference</i>.</p> <p>The bidirectional stream maintains the connection until either the client closes it or the model indicates completion. Each request and response in the stream is sent as an event with optional headers for data type and completion state.</p> <note> <p>Endpoints are scoped to an individual account, and are not public. The URL does not contain the account ID, but Amazon SageMaker AI determines the account ID from the authentication token that is supplied by the caller.</p> </note>
Args:
endpoint_name: <p>The name of the endpoint to invoke.</p>
body: <p>The request payload stream.</p>
target_variant: <p>Target variant for the request.</p>
model_invocation_path: <p>Model invocation path.</p>
model_query_string: <p>Model query string.</p>
Raises:
capo_sagemaker_runtime_http2.errors.input_validation_error.InputValidationError: <p>The input fails to satisfy the constraints specified by an AWS service.</p>
capo_sagemaker_runtime_http2.errors.internal_server_error.InternalServerError: <p>The request processing has failed because of an unknown error, exception or failure.</p>
capo_sagemaker_runtime_http2.errors.internal_stream_failure.InternalStreamFailure: <p>Internal stream failure that occurs during streaming.</p>
capo_sagemaker_runtime_http2.errors.model_error.ModelError: <p>An error occurred while processing the model.</p>
capo_sagemaker_runtime_http2.errors.model_stream_error.ModelStreamError: <p>Model stream error that occurs during streaming.</p>
capo_sagemaker_runtime_http2.errors.service_unavailable_error.ServiceUnavailableError: <p>The request has failed due to a temporary failure of the server.</p>
capo_sagemaker_runtime_http2.errors.UnknownServiceError: The service returned an error code this client does not model.
"""
async def _handler(
req: "AsyncOperationRequest[capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_input.InvokeEndpointWithBidirectionalStreamInput]",
) -> AsyncOperationResponse[
"capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_output.InvokeEndpointWithBidirectionalStreamOutput"
]:
import capo_sagemaker_runtime_http2._operations.amazon_sage_maker_runtime_http2.invoke_endpoint_with_bidirectional_stream
(
output,
http_response,
) = await capo_sagemaker_runtime_http2._operations.amazon_sage_maker_runtime_http2.invoke_endpoint_with_bidirectional_stream.async_invoke_endpoint_with_bidirectional_stream(
req.options, req.input
)
return AsyncOperationResponse(output=output, response=http_response)
def _redact_input(
input_: "capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_input.InvokeEndpointWithBidirectionalStreamInput",
) -> dict[str, Any]:
import capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_input
return capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_input.redact_sensitive(
input_
)
def _redact_output(
output: "capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_output.InvokeEndpointWithBidirectionalStreamOutput",
) -> dict[str, Any]:
import capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_output
return capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_output.redact_sensitive(
output
)
interceptors_, options_ = self.operation_options(
config_overrides,
OperationLogging(
"InvokeEndpointWithBidirectionalStream",
redact_input=_redact_input,
redact_output=_redact_output,
),
)
input_: capo_sagemaker_runtime_http2.types.invoke_endpoint_with_bidirectional_stream_input.InvokeEndpointWithBidirectionalStreamInput = {
"endpoint_name": endpoint_name,
"body": ensure_async_iterator(body),
}
if target_variant is not None:
input_["target_variant"] = target_variant
if model_invocation_path is not None:
input_["model_invocation_path"] = model_invocation_path
if model_query_string is not None:
input_["model_query_string"] = model_query_string
response = await aexecute_pipeline(
AsyncOperationRequest(input=input_, options=options_),
handler=_handler,
interceptors=list(interceptors_),
)
try:
yield response.output
finally:
await response.response.aclose()
|