Invoke a Deepgram SageMaker Endpoint

import asyncio
import json
import boto3
from aws_sdk_sagemaker_runtime_http2.client import AsyncSageMakerRuntimeHTTP2Client
from aws_sdk_sagemaker_runtime_http2.config import AsyncSageMakerRuntimeHTTP2Config
from aws_sdk_sagemaker_runtime_http2.models import (
    InvokeEndpointWithBidirectionalStreamInput,
    RequestPayloadPart,
    RequestStreamEventPayloadPart,
    ResponseStreamEventPayloadPart,
)
from smithy_http.aio.crt import AWSCRTHTTPClient
REGION = "us-east-2"
async def main():
    creds = boto3.Session().get_credentials().get_frozen_credentials()
    client = AsyncSageMakerRuntimeHTTP2Client(
        config=AsyncSageMakerRuntimeHTTP2Config(
            region=REGION,
            endpoint_uri=f"https://runtime.sagemaker.{REGION}.amazonaws.com:8443",
            aws_access_key_id=creds.access_key,
            aws_secret_access_key=creds.secret_key,
            aws_session_token=creds.token,
            transport=AWSCRTHTTPClient(),
        )
    )
    stream = await client.invoke_endpoint_with_bidirectional_stream(
        InvokeEndpointWithBidirectionalStreamInput(
            endpoint_name="<your-endpoint-name>",
            model_invocation_path="v1/listen",
            model_query_string="model=nova-3&language=en&encoding=linear16&sample_rate=16000",
        )
    )
    _, output = await stream.await_output()
    async def send(data: bytes, data_type: str):  # audio: "BINARY"; JSON control: "UTF8"
        await stream.input_stream.send(
            RequestStreamEventPayloadPart(
                value=RequestPayloadPart(bytes_=data, data_type=data_type)
            )
        )
    async def send_audio():
        with open("audio.raw", "rb") as f:
            while chunk := f.read(3200):  # 100 ms of 16 kHz linear16 audio
                await send(chunk, "BINARY")
                await asyncio.sleep(0.1)
        await send(json.dumps({"type": "CloseStream"}).encode(), "UTF8")
    async def receive_results():
        while (event := await output.receive()) is not None:
            if isinstance(event, ResponseStreamEventPayloadPart):
                print(event.value.bytes_.decode())  # Deepgram JSON transcript result
            else:  # ModelStreamError / InternalStreamFailure
                print("stream error:", event.value)
    await asyncio.gather(send_audio(), receive_results())
    await client.close()
asyncio.run(main())

import {
SageMakerRuntimeHTTP2Client,
InvokeEndpointWithBidirectionalStreamCommand,
} from "@aws-sdk/client-sagemaker-runtime-http2";

const region = "us-east-2";
const client = new SageMakerRuntimeHTTP2Client({
region,
endpoint: `https://runtime.sagemaker.${region}.amazonaws.com:8443`,
});

// Async generator yielding audio chunks (BINARY) and control messages (UTF8)
async function* requestStream() {
// yield { PayloadPart: { Bytes: audioChunk, DataType: "BINARY" } };
// yield { PayloadPart: { Bytes: new TextEncoder().encode(
//           JSON.stringify({ type: "CloseStream" })), DataType: "UTF8" } };
}

const command = new InvokeEndpointWithBidirectionalStreamCommand({
EndpointName: "<your-endpoint-name>",
ModelInvocationPath: "v1/listen",
ModelQueryString: "model=nova-3&language=en&smart_format=true",
Body: requestStream(),
});

const response = await client.send(command);

for await (const event of response.Body) {
if (event.PayloadPart?.Bytes) {
const message = new TextDecoder().decode(event.PayloadPart.Bytes);
// message is a Deepgram JSON transcript result
}
}