Invoke a Deepgram SageMaker Endpoint

Once your endpoint is deployed and in service, you invoke it to transcribe audio. A real-time endpoint supports two invocation modes, depending on how you need the response returned.

Complete, runnable examples for both modes — in Python, TypeScript, and Java — are maintained in the deepgram-devs/dg-sagemaker repository. The sections below explain each mode and link to the corresponding example. See the repository’s README for setup and prerequisites.

Use the Deepgram SDKs with the SageMaker transport

You don’t have to call the AWS APIs directly. The Deepgram SDKs can target a SageMaker endpoint through a SageMaker transport, so you keep the same client-side request and response patterns whether you call the Deepgram-hosted API or your own SageMaker deployment. You swap the transport; your listen request and result-handling code stays the same.

For example, the Deepgram Java SDK pairs with the Deepgram SageMaker transport (com.deepgram:deepgram-sagemaker):

import com.deepgram.DeepgramClient;
import com.deepgram.sagemaker.SageMakerConfig;
import com.deepgram.sagemaker.SageMakerTransportFactory;
import com.deepgram.resources.listen.v1.websocket.V1WebSocketClient;
SageMakerConfig smConfig = SageMakerConfig.builder()
        .endpointName("<your-endpoint-name>")
        .region("us-east-2")
        .build();
DeepgramClient client = DeepgramClient.builder()
        .apiKey("unused") // auth is AWS SigV4 via the transport, not a Deepgram API key
        .transportFactory(new SageMakerTransportFactory(smConfig))
        .build();
// Same SDK surface as the Deepgram-hosted API:
V1WebSocketClient ws = client.listen().v1().v1WebSocket();
ws.onResults(r -> { /* handle transcript */ });
ws.connect(connectOptions).get();
ws.sendMedia(ByteString.of(audioChunk));
// ... send a CloseStream message when finished

The remaining sections show the underlying AWS APIs directly, which apply to any language.

Streaming (real-time)

Use streaming for live, interactive transcription over a persistent bidirectional connection. You send audio chunks and receive transcription results as the audio is processed, up to 30 minutes per connection.

Streaming uses the HTTP/2 bidirectional streaming client (@aws-sdk/client-sagemaker-runtime-http2 in TypeScript, aws-sdk-sagemaker-runtime-http2 in Python) against the SageMaker bidirectional runtime endpoint (https://runtime.sagemaker.<region>.amazonaws.com:8443). The request Body is an async iterable of payload parts:

• Binary audio is sent as a Bytes payload with DataType: "BINARY".
• Control messages (for example, KeepAlive and CloseStream) are sent as UTF-8 encoded JSON with DataType: "UTF8".

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

For the complete examples — file and microphone capture, payload wrapping, keepalive handling, and stream processing — see:

• TypeScript: js-stt/stt.file.ts and stt.microphone.ts
• Python: python-stt/stt_wav_stress.py (stream subcommand)

Synchronous (real-time)

Use synchronous invocation to transcribe a single pre-recorded file and receive the full transcript in one immediate response. This is Deepgram’s “batch” transcription on a real-time endpoint — there is no streaming connection and no queue. The request body is capped at 25 MB; use streaming for larger audio.

You send the audio as the request body to InvokeEndpoint, pass the Deepgram parameters via CustomAttributes, and parse the transcript from the JSON response.

import json
import boto3
runtime = boto3.client("sagemaker-runtime", region_name="us-east-2")
with open("audio.wav", "rb") as f:
    response = runtime.invoke_endpoint(
        EndpointName="<your-endpoint-name>",
        ContentType="audio/wav",
        Accept="application/json",
        CustomAttributes="v1/listen?model=nova-3&language=en&punctuate=true",
        Body=f.read(),
    )
result = json.loads(response["Body"].read())
transcript = result["results"]["channels"][0]["alternatives"][0]["transcript"]

For the complete example, see python-stt/stt_wav_stress.py (batch subcommand) in the repository.

Asynchronous

Asynchronous invocation (InvokeEndpointAsync, files up to 1 GB) is temporarily not supported for Marketplace-hosted Deepgram. If your use case needs it, contact a Deepgram representative.

Related resources

• Validate a Deepgram SageMaker Endpoint
• Troubleshooting
• Auto-Scaling SageMaker Endpoints
• deepgram-devs/dg-sagemaker example repository