SageMaker AI Endpoint Usage Patterns

Common ways SageMaker AI endpoints are called from application code, with the deployment patterns that support them.

All examples assume:

import boto3
import json
import time

Calling an endpoint

Real-time product recommendations

An API Gateway request reaches a Lambda function, which passes the user’s context to the endpoint and returns the recommendations.

# API Gateway endpoint calling a Lambda function
def lambda_handler(event, context):
    runtime = boto3.client('sagemaker-runtime')
    user_data = {
        "user_id": event['user_id'],
        "recent_views": event['product_history'],
        "cart_items": event['cart']
    }

    response = runtime.invoke_endpoint(
        EndpointName='recommendation-endpoint',
        ContentType='application/json',
        Body=json.dumps(user_data)
    )

    recommendations = json.loads(response['Body'].read())
    return {
        'statusCode': 200,
        'body': recommendations
    }

Transaction screening

Each transaction is scored and routed by the score. Note that this loops one request at a time; for a whole file of transactions, a batch transform job is cheaper than an endpoint.

# Real-time transaction screening
def process_transaction(transaction_data):
    runtime = boto3.client('sagemaker-runtime')

    for transaction in transaction_data:
        response = runtime.invoke_endpoint(
            EndpointName='fraud-detection-endpoint',
            ContentType='application/json',
            Body=json.dumps(transaction)
        )
        risk_score = json.loads(response['Body'].read())

        if risk_score > 0.8:
            flag_for_review(transaction)
        else:
            approve_transaction(transaction)

Image content moderation

Binary payloads are sent directly, with the content type set to match.

def moderate_content(image_bytes):
    runtime = boto3.client('sagemaker-runtime')

    response = runtime.invoke_endpoint(
        EndpointName='content-moderation-endpoint',
        ContentType='application/x-image',
        Body=image_bytes
    )

    results = json.loads(response['Body'].read())
    return {
        'is_safe': results['safe_score'] > 0.95,
        'categories': results['detected_categories']
    }

Routing support tickets

# Auto-routing customer inquiries
def route_customer_ticket(ticket_text):
    runtime = boto3.client('sagemaker-runtime')

    response = runtime.invoke_endpoint(
        EndpointName='ticket-classification-endpoint',
        ContentType='application/json',
        Body=json.dumps({"text": ticket_text})
    )

    classification = json.loads(response['Body'].read())
    return assign_to_department(classification['department'])

Deployment patterns

Multi-model endpoints

Many models behind one endpoint, loaded on demand. Where the alternative is dozens of lightly used endpoints each billing continuously, this is the largest saving available.

# Cost-effective hosting of multiple models
runtime = boto3.client('sagemaker-runtime')
response = runtime.invoke_endpoint(
    EndpointName='multi-model-endpoint',
    TargetModel='customer-churn-model-v2',
    ContentType='application/json',
    Body=json.dumps(customer_data)
)

Autoscaling

Register the endpoint variant as a scalable target, then attach a target-tracking policy on invocations per instance.

# Setting up auto-scaling for endpoints
client = boto3.client('application-autoscaling')

response = client.register_scalable_target(
    ServiceNamespace='sagemaker',
    ResourceId=f'endpoint/{endpoint_name}/variant/AllTraffic',
    ScalableDimension='sagemaker:variant:DesiredInstanceCount',
    MinCapacity=1,
    MaxCapacity=4
)

# Define scaling policy based on endpoint invocations
response = client.put_scaling_policy(
    PolicyName='RequestScaling',
    ServiceNamespace='sagemaker',
    ResourceId=f'endpoint/{endpoint_name}/variant/AllTraffic',
    ScalableDimension='sagemaker:variant:DesiredInstanceCount',
    PolicyType='TargetTrackingScaling',
    TargetTrackingScalingPolicyConfiguration={
        'TargetValue': 70.0,
        'PredefinedMetricSpecification': {
            'PredefinedMetricType': 'SageMakerVariantInvocationsPerInstance'
        }
    }
)

A/B testing with production variants

Two models behind one endpoint, with traffic split by weight. Adjust the weights to shift traffic gradually rather than cutting over.

# Creating endpoint with multiple variants
create_endpoint_config_response = sagemaker_client.create_endpoint_config(
    EndpointConfigName=endpoint_config_name,
    ProductionVariants=[{
        'VariantName': 'ModelA',
        'ModelName': 'model-a',
        'InitialInstanceCount': 1,
        'InstanceType': 'ml.m5.xlarge',
        'InitialVariantWeight': 0.5
    }, {
        'VariantName': 'ModelB',
        'ModelName': 'model-b',
        'InitialInstanceCount': 1,
        'InstanceType': 'ml.m5.xlarge',
        'InitialVariantWeight': 0.5
    }]
)

Operational patterns

Custom prediction metrics

SageMaker AI publishes invocation and latency metrics automatically. Prediction quality is application-specific and has to be published explicitly, once ground truth becomes available.

# CloudWatch custom metrics for model monitoring
cloudwatch = boto3.client('cloudwatch')

def log_prediction_metrics(prediction, actual):
    cloudwatch.put_metric_data(
        Namespace='ModelMetrics',
        MetricData=[{
            'MetricName': 'PredictionAccuracy',
            'Value': calculate_accuracy(prediction, actual),
            'Unit': 'Percent'
        }]
    )

Retry with exponential backoff

def invoke_endpoint_with_retry(endpoint_name, payload, max_retries=3):
    runtime = boto3.client('sagemaker-runtime')

    for attempt in range(max_retries):
        try:
            response = runtime.invoke_endpoint(
                EndpointName=endpoint_name,
                ContentType='application/json',
                Body=json.dumps(payload)
            )
            return json.loads(response['Body'].read())
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # Exponential backoff

Catching bare Exception retries programming errors as well as transient ones. In production, narrow this to the botocore exceptions worth retrying — throttling and service-unavailable responses — and let the rest fail immediately.

What these patterns have in common

• Endpoints bill while they exist, so match the deployment shape to the traffic: batch transform for whole datasets, serverless inference for intermittent traffic, multi-model endpoints for many small models.
• Publish the metrics that describe prediction quality, not just the ones the service gives you for free.
• Handle failure explicitly at the call site — an endpoint is a network dependency like any other.
• Use production variants to shift traffic between model versions rather than replacing an endpoint in place.